feat(v1.8.0): RSS leak fix, parallel snapshot deserialize, replication WAL origin

Three production wins motivated by the Zoe incident (OOM every ~3h on
1.7.5) and a user-reported "100% CPU on one core during boot with much
data" complaint.

--- 1. RSS leak: drop version history on eviction ---

Root cause: `saveToHistory()` pushes the prior version into
`versionHistory[id]` on every update, but `maxVersions=0` (unlimited)
on hot-write collections like `messages`, `documents`, `kb_articles`,
`kb_chunks` lets the deque grow without bound. Eviction removed the
live document from `documents[]` and subtracted only the doc's bytes
from `estimatedMemoryBytes_` — the orphaned history deque stayed
pinned to the heap forever. Under heavy LLM streaming (every
appendContent/appendThinking is an update) plus cron jobs on Zoe,
the leak walked RSS to 11 GB against a 6 GB cap in ~30 min and the
kernel killed the process.

Fix: in `evictDocument`, also erase `versionHistory[id]` and subtract
the freed history bytes from the budget. The existing hot-load path
(`loadEvictedDocument`) already didn't restore history, so dropping
it on eviction is consistent with the user-observable contract.

--- 2. Boot perf: parallel snapshot deserialize ---

Pre-v1.8 `SnapshotManager::deserializeStore` was a single sequential
JSON parse over every document — for a 5 GB snapshot on shadowman this
pinned a single core at 100% for tens of seconds. The body is a
concatenation of independent per-collection blocks, so we now:
  1. Walk the body once with pure byte arithmetic to record each
     collection's byte ranges (cheap, stays in L2).
  2. Create the collections sequentially so global state is consistent
     before workers start.
  3. Spawn N=hardware_concurrency() worker threads that pull from a
     shared atomic index and parse + load each collection's payload in
     isolation. Per-collection mutexes mean no two workers contend on
     the same collection's documents map.

For shadowman-shaped workloads (~30 collections) this drops boot
roughly linearly with cores. Single-giant-collection workloads still
serialize within the collection — a follow-up.

--- 3. `loadDocument` is now a proper upsert ---

Pre-v1.8 `MemoryStore::loadDocument` unconditionally bumped
`stats_.totalDocuments++` and `estimatedMemoryBytes_.fetch_add(docSize)`
even when overwriting an existing entry. With replication delivering
the same entry twice (once via SyncStream push and once via
GetEntriesSince pull during the same window), this double-counted —
inflated the budget, forced premature eviction, and produced the
"eviction: nothing evictable, pressure=emergency" stall the
replica-eviction load test now catches. Fix: compute the size delta
vs. the previous occupant.

--- 4. GetEntriesSince preserves origin nodeId ---

`database_grpc_impl.cpp` no longer hard-codes the responder's nodeId
into the streamed ReplicationEntry; it now uses `walEntry.nodeId` and
falls back to the local node id only when an empty (pre-v1.8) WAL
entry is encountered. Pairs with the SyncProtocol "skip own writes"
check below — together they break the replication echo amplification
the v1.7.2 draft hit.

--- 5. SyncProtocol skips entries originating from the local node ---

`pullEntriesFromPeer` drops entries where `entry.node_id() ==
config_.nodeId`. Without this, a write we produced would be shipped
back to us via the pull path, re-applied, re-broadcast, and ping-pong
forever (this is exactly what blew up the v1.7.2 simple-fix draft:
5000 inserts → 216k WAL entries on the follower).

--- Acceptance status ---

- `test_eviction`, `test_snapshot_durability`, `test_vector_storage`,
  `test_timestamp_precision` all pass.
- Replica-eviction load test: 0/5000 (baseline 1.7.5) → 998/5000 with
  this commit. Significant progress, not yet 100% — the WAL contains
  all 5000 entries (and duplicates from push+pull) but
  `persistence_->loadDocument` returns nullopt for ~4000 of them under
  the read storm. Tracked as a v1.8.1 follow-up; the production-blocking
  symptoms (Zoe OOM, slow boot) are fixed here.

--- Packaging ---

- Build dep `libsystemd-dev` and runtime dep `libsystemd0` so the
  `sd_notify(READY=1)` path from v1.7.5 links cleanly in the deb image.

VERSION

@@ -1 +1 @@
-1.7.5
+1.8.0

packaging/Dockerfile.base

@@ -17,6 +17,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     ccache \
     libssl-dev \
     liblz4-dev \
+    libsystemd-dev \
     protobuf-compiler \
     libprotobuf-dev \
     libgrpc++-dev \
@@ -34,4 +35,4 @@ ENV CCACHE_COMPRESSLEVEL=6
 
 WORKDIR /build
 
-RUN echo "smartbotic-db-build-base:debian13:v1" > /etc/smartbotic-db-build-base
+RUN echo "smartbotic-db-build-base:debian13:v2" > /etc/smartbotic-db-build-base

packaging/deb/templates/control.server

@@ -14,5 +14,5 @@ License: proprietary
 Provides: shadowman-database, callerai-storage
 Conflicts: shadowman-database, callerai-storage
 Replaces: shadowman-database, callerai-storage
-Depends: libsmartbotic-db-client (= {{VERSION}}), libssl3t64, liblz4-1
+Depends: libsmartbotic-db-client (= {{VERSION}}), libssl3t64, liblz4-1, libsystemd0
 Installed-Size: {{INSTALLED_SIZE}}

service/src/database_grpc_impl.cpp

@@ -1558,7 +1558,10 @@ grpc::Status DatabaseReplicationGrpcImpl::GetEntriesSince(
             pb::ReplicationEntry entry;
             entry.set_sequence(walEntry.sequence);
             entry.set_global_timestamp(walEntry.timestamp);
-            entry.set_node_id(nodeState.nodeId);
+            // v1.8.0 — preserve the original origin nodeId from the WAL.
+            // Pre-v1.8 entries have an empty walEntry.nodeId; fall back to
+            // the local node id for those so peers see something stable.
+            entry.set_node_id(walEntry.nodeId.empty() ? nodeState.nodeId : walEntry.nodeId);
             entry.set_collection(walEntry.collection);
             entry.set_document_id(walEntry.documentId);
 

service/src/memory_store.cpp

@@ -1911,6 +1911,27 @@ void MemoryStore::loadDocument(const std::string& collection, Document doc) {
     // Calculate size before moving
     uint64_t docSize = estimateDocumentSize(doc);
 
+    // v1.8.0 — loadDocument is now a proper upsert with respect to stats
+    // and the memory budget. Pre-v1.8 it unconditionally bumped both even
+    // when overwriting an existing entry. Replication apply (push + pull)
+    // and WAL replay both reach this path, and a single doc can land twice
+    // — once via SyncStream broadcast and once via GetEntriesSince pull —
+    // which inflated `estimatedMemoryBytes_` and `stats_.totalDocuments`,
+    // pushed the budget to emergency, and ultimately produced the
+    // "evictable: 0 docs" deadlock seen in the replica-eviction load test.
+    auto existing = coll->documents.find(doc.id);
+    uint64_t oldSize = 0;
+    bool replaced = false;
+    if (existing != coll->documents.end()) {
+        oldSize = estimateDocumentSize(existing->second);
+        // Remove the prior expiration-index entry — the new doc may have a
+        // different expiresAt (or none).
+        if (existing->second.expiresAt > 0) {
+            removeFromExpirationIndex(*coll, doc.id, existing->second.expiresAt);
+        }
+        replaced = true;
+    }
+
     // Add to expiration index
     if (doc.expiresAt > 0) {
         addToExpirationIndex(*coll, doc.id, doc.expiresAt);
@@ -1918,13 +1939,17 @@ void MemoryStore::loadDocument(const std::string& collection, Document doc) {
 
     coll->documents[doc.id] = std::move(doc);
 
-    {
+    if (!replaced) {
         std::lock_guard<std::mutex> statsLock(statsMutex_);
         stats_.totalDocuments++;
     }
 
-    // Update memory tracking
-    estimatedMemoryBytes_.fetch_add(docSize, std::memory_order_relaxed);
+    // Update memory tracking: bump only the delta vs. the previous occupant.
+    if (docSize > oldSize) {
+        estimatedMemoryBytes_.fetch_add(docSize - oldSize, std::memory_order_relaxed);
+    } else if (oldSize > docSize) {
+        estimatedMemoryBytes_.fetch_sub(oldSize - docSize, std::memory_order_relaxed);
+    }
 }
 
 void MemoryStore::loadDocumentWithHistory(const std::string& collection, Document doc) {
@@ -2854,6 +2879,24 @@ bool MemoryStore::evictDocument(const std::string& collection, const std::string
     // Remove document from memory
     coll->documents.erase(it);
 
+    // v1.8.0 — also drop the doc's version history. Before this, eviction only
+    // freed the live document but left `versionHistory[id]` intact, which on
+    // hot-write collections with `max_versions: 0` (e.g. ShadowMan `messages`,
+    // `documents`, `kb_*`) caused unbounded growth: every update pushed a
+    // copy of the prior version into the deque, eviction never touched it,
+    // and the orphaned deque stayed pinned to the heap forever. See
+    // BUG-memory-leak-zoe.md. Hot-load via `loadEvictedDocument` already
+    // doesn't restore history, so dropping it here is a no-op vs the
+    // pre-v1.8 user-observable contract.
+    uint64_t historyBytesFreed = 0;
+    auto histIt = coll->versionHistory.find(id);
+    if (histIt != coll->versionHistory.end()) {
+        for (const auto& ver : histIt->second) {
+            historyBytesFreed += estimateDocumentVersionSize(ver);
+        }
+        coll->versionHistory.erase(histIt);
+    }
+
     collLock.unlock();
 
     // Store stub in evicted map
@@ -2870,10 +2913,12 @@ bool MemoryStore::evictDocument(const std::string& collection, const std::string
         stats_.evictionCount++;
     }
 
-    // Update memory tracking atomically (subtract evicted document size)
-    estimatedMemoryBytes_.fetch_sub(stub.estimatedSize, std::memory_order_relaxed);
+    // Update memory tracking atomically (subtract evicted document + history)
+    estimatedMemoryBytes_.fetch_sub(stub.estimatedSize + historyBytesFreed,
+                                    std::memory_order_relaxed);
 
-    SPDLOG_DEBUG("Evicted document {}/{} (size: {} bytes)", collection, id, stub.estimatedSize);
+    SPDLOG_DEBUG("Evicted document {}/{} (doc: {} bytes, history: {} bytes)",
+                 collection, id, stub.estimatedSize, historyBytesFreed);
 
     return true;
 }

service/src/persistence/snapshot.cpp

@@ -2,6 +2,7 @@
 #include "wal.hpp"
 
 #include <algorithm>
+#include <atomic>
 #include <cerrno>
 #include <chrono>
 #include <cstring>
@@ -9,6 +10,7 @@
 #include <iomanip>
 #include <sstream>
 #include <system_error>
+#include <thread>
 
 #include <fcntl.h>
 #include <unistd.h>
@@ -728,159 +730,279 @@ std::vector<uint8_t> SnapshotManager::serializeStore(const MemoryStore& store,
     return result;
 }
 
+namespace {
+
+// v1.8.0 — slice describing where one collection's bytes live in the
+// decompressed snapshot body, used by the two-phase parallel deserializer.
+struct CollectionSlice {
+    std::string name;
+    CollectionOptions options;
+    uint64_t docCount = 0;
+    size_t docSectionStart = 0;
+    size_t docSectionEnd = 0;
+    uint64_t historyEntryCount = 0;
+    size_t historySectionStart = 0;
+    size_t historySectionEnd = 0;
+    uint64_t vecCount = 0;
+    size_t vecSectionStart = 0;
+    size_t vecSectionEnd = 0;
+};
+
+}  // namespace
+
 void SnapshotManager::deserializeStore(const std::vector<uint8_t>& data, MemoryStore& store,
                                        uint32_t snapshotVersion) const {
+    // v1.8.0 — two-phase parallel deserializer. Before this, a 5 GB snapshot
+    // with many JSON-encoded documents pinned a single core at 100% for tens
+    // of seconds during boot. The serialized format is a concatenation of
+    // independent per-collection blocks, so we:
+    //   1. Walk the body once with pure byte arithmetic (no JSON parsing) to
+    //      record each collection's name, options, and the byte ranges that
+    //      hold its docs / history / vectors. This is cheap and stays in L2.
+    //   2. Create the collections sequentially so global state (collection
+    //      map, options) is consistent before workers start.
+    //   3. Spawn N worker threads (N = hardware_concurrency capped at the
+    //      number of collections) that pull work off a shared index counter
+    //      and parse + load each collection's payload in isolation. JSON
+    //      parsing — the actual hot path — runs in parallel; per-collection
+    //      MemoryStore mutexes mean no two workers contend on the same
+    //      collection's documents map.
+    //
+    // For workloads with many small/medium collections (e.g. ShadowMan, with
+    // ~30 collections) this drops boot time roughly linearly with cores. For
+    // a single-giant-collection workload there is no within-collection
+    // parallelism yet; that would be a follow-up.
+
+    // ===== Phase 1: scan body byte-by-byte to find collection slices =====
+    std::vector<CollectionSlice> slices;
     size_t offset = 0;
-
     while (offset < data.size()) {
-        // Read collection name
+        CollectionSlice slice;
+
+        // Collection name
         if (offset + 2 > data.size()) break;
         uint16_t nameLen = static_cast<uint16_t>(data[offset]) |
                           (static_cast<uint16_t>(data[offset + 1]) << 8);
         offset += 2;
-
         if (offset + nameLen > data.size()) break;
-        std::string collName(reinterpret_cast<const char*>(&data[offset]), nameLen);
+        slice.name.assign(reinterpret_cast<const char*>(&data[offset]), nameLen);
         offset += nameLen;
 
-        // Read collection options
+        // Collection options (small JSON — keep parsing here, sequential)
         if (offset + 4 > data.size()) break;
         uint32_t optionsLen = 0;
         for (int i = 0; i < 4; ++i) {
             optionsLen |= static_cast<uint32_t>(data[offset++]) << (i * 8);
         }
-
         if (offset + optionsLen > data.size()) break;
         std::string optionsJson(reinterpret_cast<const char*>(&data[offset]), optionsLen);
         offset += optionsLen;
-
-        CollectionOptions options;
         try {
-            options = CollectionOptions::fromJson(nlohmann::json::parse(optionsJson));
+            slice.options = CollectionOptions::fromJson(nlohmann::json::parse(optionsJson));
         } catch (const nlohmann::json::exception&) {
-            // Use default options
+            // Defaults — same fallback as the pre-v1.8 path.
         }
 
-        // Create collection
-        store.createCollection(collName, options);
-
-        // Read document count
+        // Document section — record its byte range, skip past without parsing
         if (offset + 8 > data.size()) break;
-        uint64_t docCount = 0;
         for (int i = 0; i < 8; ++i) {
-            docCount |= static_cast<uint64_t>(data[offset++]) << (i * 8);
+            slice.docCount |= static_cast<uint64_t>(data[offset++]) << (i * 8);
         }
-
-        // Read each document
-        for (uint64_t i = 0; i < docCount; ++i) {
-            if (offset + 4 > data.size()) break;
+        slice.docSectionStart = offset;
+        for (uint64_t i = 0; i < slice.docCount; ++i) {
+            if (offset + 4 > data.size()) { offset = data.size(); break; }
             uint32_t docLen = 0;
             for (int j = 0; j < 4; ++j) {
                 docLen |= static_cast<uint32_t>(data[offset++]) << (j * 8);
             }
-
-            if (offset + docLen > data.size()) break;
-            std::string docJson(reinterpret_cast<const char*>(&data[offset]), docLen);
+            if (offset + docLen > data.size()) { offset = data.size(); break; }
             offset += docLen;
-
-            try {
-                Document doc = Document::fromJson(nlohmann::json::parse(docJson));
-                store.loadDocument(collName, doc);
-            } catch (const nlohmann::json::exception&) {
-                // Skip invalid document
-            }
         }
+        slice.docSectionEnd = offset;
 
-        // Read version history (v2+)
+        // History section (v2+)
         if (snapshotVersion >= 2) {
             if (offset + 8 > data.size()) break;
-            uint64_t historyEntryCount = 0;
             for (int i = 0; i < 8; ++i) {
-                historyEntryCount |= static_cast<uint64_t>(data[offset++]) << (i * 8);
+                slice.historyEntryCount |= static_cast<uint64_t>(data[offset++]) << (i * 8);
             }
-
-            for (uint64_t h = 0; h < historyEntryCount; ++h) {
-                // Read doc ID
-                if (offset + 2 > data.size()) break;
+            slice.historySectionStart = offset;
+            for (uint64_t h = 0; h < slice.historyEntryCount; ++h) {
+                if (offset + 2 > data.size()) { offset = data.size(); break; }
                 uint16_t docIdLen = static_cast<uint16_t>(data[offset]) |
                                    (static_cast<uint16_t>(data[offset + 1]) << 8);
                 offset += 2;
-
-                if (offset + docIdLen > data.size()) break;
-                std::string docId(reinterpret_cast<const char*>(&data[offset]), docIdLen);
+                if (offset + docIdLen > data.size()) { offset = data.size(); break; }
                 offset += docIdLen;
-
-                // Read version count
-                if (offset + 4 > data.size()) break;
+                if (offset + 4 > data.size()) { offset = data.size(); break; }
                 uint32_t versionCount = 0;
                 for (int i = 0; i < 4; ++i) {
                     versionCount |= static_cast<uint32_t>(data[offset++]) << (i * 8);
                 }
-
-                // Read each version into a deque
-                std::deque<DocumentVersion> history;
                 for (uint32_t v = 0; v < versionCount; ++v) {
-                    if (offset + 4 > data.size()) break;
+                    if (offset + 4 > data.size()) { offset = data.size(); break; }
                     uint32_t verLen = 0;
                     for (int i = 0; i < 4; ++i) {
                         verLen |= static_cast<uint32_t>(data[offset++]) << (i * 8);
                     }
-
-                    if (offset + verLen > data.size()) break;
-                    std::string verJson(reinterpret_cast<const char*>(&data[offset]), verLen);
+                    if (offset + verLen > data.size()) { offset = data.size(); break; }
                     offset += verLen;
-
-                    try {
-                        auto ver = DocumentVersion::fromJson(nlohmann::json::parse(verJson));
-                        history.push_back(std::move(ver));
-                    } catch (const nlohmann::json::exception&) {
-                        // Skip invalid version entry
-                    }
-                }
-
-                if (!history.empty()) {
-                    store.loadVersionHistory(collName, docId, std::move(history));
                 }
             }
+            slice.historySectionEnd = offset;
         }
 
-        // Read vector data (v3+)
+        // Vector section (v3+)
         if (snapshotVersion >= 3) {
             if (offset + 8 > data.size()) break;
-            uint64_t vecCount = 0;
             for (int i = 0; i < 8; ++i) {
-                vecCount |= static_cast<uint64_t>(data[offset++]) << (i * 8);
+                slice.vecCount |= static_cast<uint64_t>(data[offset++]) << (i * 8);
             }
-
-            for (uint64_t v = 0; v < vecCount; ++v) {
-                // Read doc ID
-                if (offset + 2 > data.size()) break;
+            slice.vecSectionStart = offset;
+            for (uint64_t v = 0; v < slice.vecCount; ++v) {
+                if (offset + 2 > data.size()) { offset = data.size(); break; }
                 uint16_t docIdLen = static_cast<uint16_t>(data[offset]) |
                                    (static_cast<uint16_t>(data[offset + 1]) << 8);
                 offset += 2;
-
-                if (offset + docIdLen > data.size()) break;
-                std::string docId(reinterpret_cast<const char*>(&data[offset]), docIdLen);
+                if (offset + docIdLen > data.size()) { offset = data.size(); break; }
                 offset += docIdLen;
-
-                // Read vector dimension
-                if (offset + 4 > data.size()) break;
+                if (offset + 4 > data.size()) { offset = data.size(); break; }
                 uint32_t dim = 0;
                 for (int i = 0; i < 4; ++i) {
                     dim |= static_cast<uint32_t>(data[offset++]) << (i * 8);
                 }
-
-                // Read raw float data
                 size_t byteCount = static_cast<size_t>(dim) * sizeof(float);
-                if (offset + byteCount > data.size()) break;
-                std::vector<float> vec(dim);
-                std::memcpy(vec.data(), &data[offset], byteCount);
+                if (offset + byteCount > data.size()) { offset = data.size(); break; }
                 offset += byteCount;
+            }
+            slice.vecSectionEnd = offset;
+        }
 
+        slices.push_back(std::move(slice));
+    }
+
+    // ===== Phase 2: create all collections sequentially =====
+    // Doing this before workers start means each worker only mutates its own
+    // collection's per-coll mutex, never the global collections map.
+    for (const auto& slice : slices) {
+        store.createCollection(slice.name, slice.options);
+    }
+
+    // ===== Phase 3: parallel parse + load per collection =====
+    // Worker: pop a slice index, parse + load all of its docs / history /
+    // vectors. JSON parsing is the dominant cost; per-collection isolation
+    // means no shared-mutex contention except on the lock-free atomics.
+    const auto loadOneSlice = [&data, &store, snapshotVersion](const CollectionSlice& slice) {
+        const std::string& collName = slice.name;
+
+        // --- Documents ---
+        size_t off = slice.docSectionStart;
+        for (uint64_t i = 0; i < slice.docCount && off < slice.docSectionEnd; ++i) {
+            if (off + 4 > slice.docSectionEnd) break;
+            uint32_t docLen = 0;
+            for (int j = 0; j < 4; ++j) {
+                docLen |= static_cast<uint32_t>(data[off++]) << (j * 8);
+            }
+            if (off + docLen > slice.docSectionEnd) break;
+            std::string docJson(reinterpret_cast<const char*>(&data[off]), docLen);
+            off += docLen;
+            try {
+                Document doc = Document::fromJson(nlohmann::json::parse(docJson));
+                store.loadDocument(collName, doc);
+            } catch (const nlohmann::json::exception&) {
+                // Skip invalid document — same tolerance as pre-v1.8.
+            }
+        }
+
+        // --- History ---
+        if (snapshotVersion >= 2) {
+            off = slice.historySectionStart;
+            for (uint64_t h = 0; h < slice.historyEntryCount && off < slice.historySectionEnd; ++h) {
+                if (off + 2 > slice.historySectionEnd) break;
+                uint16_t docIdLen = static_cast<uint16_t>(data[off]) |
+                                   (static_cast<uint16_t>(data[off + 1]) << 8);
+                off += 2;
+                if (off + docIdLen > slice.historySectionEnd) break;
+                std::string docId(reinterpret_cast<const char*>(&data[off]), docIdLen);
+                off += docIdLen;
+                if (off + 4 > slice.historySectionEnd) break;
+                uint32_t versionCount = 0;
+                for (int i = 0; i < 4; ++i) {
+                    versionCount |= static_cast<uint32_t>(data[off++]) << (i * 8);
+                }
+                std::deque<DocumentVersion> history;
+                for (uint32_t v = 0; v < versionCount; ++v) {
+                    if (off + 4 > slice.historySectionEnd) break;
+                    uint32_t verLen = 0;
+                    for (int i = 0; i < 4; ++i) {
+                        verLen |= static_cast<uint32_t>(data[off++]) << (i * 8);
+                    }
+                    if (off + verLen > slice.historySectionEnd) break;
+                    std::string verJson(reinterpret_cast<const char*>(&data[off]), verLen);
+                    off += verLen;
+                    try {
+                        auto ver = DocumentVersion::fromJson(nlohmann::json::parse(verJson));
+                        history.push_back(std::move(ver));
+                    } catch (const nlohmann::json::exception&) {
+                        // Skip invalid version entry.
+                    }
+                }
+                if (!history.empty()) {
+                    store.loadVersionHistory(collName, docId, std::move(history));
+                }
+            }
+        }
+
+        // --- Vectors ---
+        if (snapshotVersion >= 3) {
+            off = slice.vecSectionStart;
+            for (uint64_t v = 0; v < slice.vecCount && off < slice.vecSectionEnd; ++v) {
+                if (off + 2 > slice.vecSectionEnd) break;
+                uint16_t docIdLen = static_cast<uint16_t>(data[off]) |
+                                   (static_cast<uint16_t>(data[off + 1]) << 8);
+                off += 2;
+                if (off + docIdLen > slice.vecSectionEnd) break;
+                std::string docId(reinterpret_cast<const char*>(&data[off]), docIdLen);
+                off += docIdLen;
+                if (off + 4 > slice.vecSectionEnd) break;
+                uint32_t dim = 0;
+                for (int i = 0; i < 4; ++i) {
+                    dim |= static_cast<uint32_t>(data[off++]) << (i * 8);
+                }
+                size_t byteCount = static_cast<size_t>(dim) * sizeof(float);
+                if (off + byteCount > slice.vecSectionEnd) break;
+                std::vector<float> vec(dim);
+                std::memcpy(vec.data(), &data[off], byteCount);
+                off += byteCount;
                 store.loadVector(collName, docId, std::move(vec));
             }
         }
+    };
+
+    if (slices.size() <= 1) {
+        // Trivial fast path — single collection, no parallelism overhead.
+        for (const auto& slice : slices) loadOneSlice(slice);
+        return;
+    }
+
+    const size_t hwc = std::thread::hardware_concurrency();
+    const size_t workerCount = std::min(slices.size(), hwc > 0 ? hwc : 4);
+    std::atomic<size_t> nextSlice{0};
+    std::vector<std::thread> workers;
+    workers.reserve(workerCount);
+    for (size_t w = 0; w < workerCount; ++w) {
+        workers.emplace_back([&]() {
+            for (;;) {
+                size_t i = nextSlice.fetch_add(1, std::memory_order_relaxed);
+                if (i >= slices.size()) break;
+                loadOneSlice(slices[i]);
+            }
+        });
     }
+    for (auto& t : workers) t.join();
+
+    spdlog::info("Snapshot deserialized: {} collections via {} workers",
+                 slices.size(), workerCount);
 }
 
 } // namespace smartbotic::database

service/src/replication/sync_protocol.cpp

@@ -305,6 +305,17 @@ void SyncProtocol::pullEntriesFromPeer(PeerConnection& peer, uint64_t fromSequen
             uint64_t lastSeq = currentSeq;
 
             while (reader->Read(&entry)) {
+                // v1.8.0 — drop entries we originally produced. The peer's
+                // GetEntriesSince now preserves origin nodeId, so a write
+                // that left this node, was applied on the peer, and is now
+                // being shipped back as a "pulled" entry shows up with our
+                // own nodeId. Applying it would re-broadcast (echo
+                // amplification, see v1.7.2 draft incident).
+                if (!config_.nodeId.empty() && entry.node_id() == config_.nodeId) {
+                    lastSeq = entry.sequence();
+                    ++count;
+                    continue;
+                }
                 if (entryCallback_) {
                     entryCallback_(entry);
                 }