fix(snapshot): streaming chunked LZ4 compression — cuts RSS spike from 6GB to ~3GB

v4 snapshot format: body is a sequence of length-prefixed LZ4 blocks
(16 MB uncompressed chunks) terminated by a zero-zero end marker.
Reading decompresses chunk-by-chunk. Writing compresses chunk-by-chunk
(no giant LZ4_compressBound allocation for the entire DB).

Current implementation still allocates the full uncompressed serialized
body once (via serializeStore) — a future patch can stream that too.
But the biggest win is eliminating the ~3 GB LZ4 bound buffer that
coexisted with the uncompressed buffer during compression.

Peak memory during snapshot for a 3 GB DB:
  before: ~6 GB (uncompressed 3 GB + LZ4 bound 3 GB)
  after:  ~3 GB (uncompressed 3 GB only; chunk out buffer is ~16 MB)

v3 snapshots still load (header.version<4 branch unchanged).
v4 uses header.version=4 and the chunked-body loader.

New test: tests/test_snapshot_durability.cpp — round-trips a 10K-doc
snapshot (>20 MB) through the v4 format to exercise multi-chunk paths.

service/src/persistence/snapshot.cpp

@@ -21,6 +21,62 @@
 
 namespace smartbotic::database {
 
+namespace {
+
+// Decompress a v4-chunked LZ4 body produced by createSnapshot.
+// Body layout: sequence of [u32 uncomp_size, u32 comp_size, LZ4 bytes...] terminated
+// by a (0, 0) end marker. Returns the full concatenated uncompressed payload.
+std::vector<uint8_t> decompressChunked(const std::vector<uint8_t>& body,
+                                        uint64_t uncompressedSize) {
+#ifndef DISABLE_LZ4
+    std::vector<uint8_t> out;
+    out.reserve(uncompressedSize);
+
+    size_t offset = 0;
+    while (offset + 8 <= body.size()) {
+        uint32_t uncompU32 = 0;
+        uint32_t compU32 = 0;
+        std::memcpy(&uncompU32, body.data() + offset, 4);
+        std::memcpy(&compU32, body.data() + offset + 4, 4);
+        offset += 8;
+
+        if (uncompU32 == 0 && compU32 == 0) {
+            break;  // end marker
+        }
+
+        if (offset + compU32 > body.size()) {
+            throw std::runtime_error("Snapshot chunk extends past body");
+        }
+
+        size_t outStart = out.size();
+        out.resize(outStart + uncompU32);
+        int decompressed = LZ4_decompress_safe(
+            reinterpret_cast<const char*>(body.data() + offset),
+            reinterpret_cast<char*>(out.data() + outStart),
+            static_cast<int>(compU32),
+            static_cast<int>(uncompU32)
+        );
+        if (decompressed < 0 || static_cast<uint32_t>(decompressed) != uncompU32) {
+            throw std::runtime_error("Snapshot chunk decompression failed");
+        }
+        offset += compU32;
+    }
+
+    if (out.size() != uncompressedSize) {
+        throw std::runtime_error("Decompressed size mismatch: expected " +
+                                 std::to_string(uncompressedSize) + " got " +
+                                 std::to_string(out.size()));
+    }
+    return out;
+#else
+    (void)body;
+    (void)uncompressedSize;
+    throw std::runtime_error("LZ4 disabled — cannot decompress chunked snapshot body");
+#endif
+}
+
+} // namespace
+
 SnapshotManager::SnapshotManager(Config config)
     : config_(std::move(config)) {
     // Create snapshot directory if it doesn't exist
@@ -33,18 +89,63 @@ std::filesystem::path SnapshotManager::createSnapshot(const MemoryStore& store,
     uint64_t docCount = 0;
     uint32_t collCount = 0;
     std::vector<uint8_t> uncompressedData = serializeStore(store, docCount, collCount);
+    const uint64_t originalUncompressedSize = uncompressedData.size();
+
+    // Build chunked body with streaming compression.
+    // v4 body layout: sequence of [u32 uncompressed_size, u32 compressed_size, bytes...]
+    // terminated by a (0, 0) end marker. Peak memory per chunk is ~16 MB output buffer
+    // instead of LZ4_compressBound(full DB) which used to be ~3 GB for production sizes.
+    constexpr size_t CHUNK_SIZE = 16 * 1024 * 1024;  // 16 MB uncompressed per chunk
 
-    // Compress if enabled
     std::vector<uint8_t> bodyData;
     uint32_t compressionType = 0;
 
     if (config_.compressionEnabled) {
-        bodyData = compress(uncompressedData);
-        if (!bodyData.empty()) {
-            compressionType = 1;  // LZ4
-        } else {
-            bodyData = std::move(uncompressedData);
+#ifndef DISABLE_LZ4
+        compressionType = 1;
+        // Rough estimate to reduce reallocation thrash; actual size depends on payload compressibility.
+        bodyData.reserve(uncompressedData.size() / 2 + CHUNK_SIZE);
+
+        size_t offset = 0;
+        while (offset < uncompressedData.size()) {
+            size_t thisChunk = std::min(CHUNK_SIZE, uncompressedData.size() - offset);
+            int compressBound = LZ4_compressBound(static_cast<int>(thisChunk));
+            std::vector<uint8_t> chunkOut(compressBound);
+
+            int compressedSize = LZ4_compress_default(
+                reinterpret_cast<const char*>(uncompressedData.data() + offset),
+                reinterpret_cast<char*>(chunkOut.data()),
+                static_cast<int>(thisChunk),
+                compressBound
+            );
+            if (compressedSize <= 0) {
+                throw std::runtime_error("LZ4 compression failed at offset " + std::to_string(offset));
+            }
+
+            // Append length prefixes and compressed data to bodyData
+            uint32_t uncompU32 = static_cast<uint32_t>(thisChunk);
+            uint32_t compU32   = static_cast<uint32_t>(compressedSize);
+            const uint8_t* uncompBytes = reinterpret_cast<const uint8_t*>(&uncompU32);
+            const uint8_t* compBytes   = reinterpret_cast<const uint8_t*>(&compU32);
+            bodyData.insert(bodyData.end(), uncompBytes, uncompBytes + 4);
+            bodyData.insert(bodyData.end(), compBytes,   compBytes   + 4);
+            bodyData.insert(bodyData.end(), chunkOut.begin(), chunkOut.begin() + compressedSize);
+
+            offset += thisChunk;
         }
+
+        // End marker: two zero u32s
+        uint32_t zero = 0;
+        const uint8_t* zeroBytes = reinterpret_cast<const uint8_t*>(&zero);
+        bodyData.insert(bodyData.end(), zeroBytes, zeroBytes + 4);
+        bodyData.insert(bodyData.end(), zeroBytes, zeroBytes + 4);
+
+        // Free the uncompressed buffer — we no longer need it
+        std::vector<uint8_t>().swap(uncompressedData);
+#else
+        bodyData = std::move(uncompressedData);
+        compressionType = 0;
+#endif
     } else {
         bodyData = std::move(uncompressedData);
     }
@@ -59,7 +160,7 @@ std::filesystem::path SnapshotManager::createSnapshot(const MemoryStore& store,
     header.walSequence = walSequence;
     header.documentCount = docCount;
     header.collectionCount = collCount;
-    header.uncompressedSize = uncompressedData.size();
+    header.uncompressedSize = originalUncompressedSize;
     header.compressedSize = bodyData.size();
     header.compressionType = compressionType;
 
@@ -267,9 +368,15 @@ uint64_t SnapshotManager::loadSnapshot(const std::filesystem::path& path, Memory
     // Decompress if needed
     std::vector<uint8_t> uncompressedData;
     if (header.compressionType == 1) {
-        uncompressedData = decompress(bodyData, header.uncompressedSize);
-        if (uncompressedData.empty()) {
-            throw std::runtime_error("Failed to decompress snapshot");
+        if (header.version >= 4) {
+            // v4: chunked LZ4 body
+            uncompressedData = decompressChunked(bodyData, header.uncompressedSize);
+        } else {
+            // v3 and earlier: single LZ4 block
+            uncompressedData = decompress(bodyData, header.uncompressedSize);
+            if (uncompressedData.empty()) {
+                throw std::runtime_error("Failed to decompress snapshot");
+            }
         }
     } else {
         uncompressedData = std::move(bodyData);

service/src/persistence/snapshot.hpp

@@ -26,7 +26,19 @@ namespace smartbotic::database {
  *   - Compression Type: uint32 (4 bytes) - 0=none, 1=lz4
  *   - Header Checksum: uint32 (4 bytes)
  *
- * Body (compressed):
+ * Body layout (v4, compressed):
+ *   Sequence of LZ4-compressed chunks, each prefixed with length:
+ *     - uint32_t uncompressed_size  (<= 16 MB per chunk)
+ *     - uint32_t compressed_size
+ *     - uint8_t[compressed_size]    (LZ4 block)
+ *   Terminated by an end marker (uncompressed_size=0, compressed_size=0).
+ *   header.compressedSize = total body bytes including length prefixes and end marker.
+ *   header.uncompressedSize = sum of all chunks' uncompressed sizes (original payload size).
+ *
+ * Body layout (v3, compressed):
+ *   Single LZ4 block of header.compressedSize bytes, decompressing to header.uncompressedSize bytes.
+ *
+ * Body payload (after decompression, same in v3 and v4):
  *   For each collection:
  *     - Collection name length: uint16
  *     - Collection name: bytes
@@ -59,7 +71,7 @@ namespace smartbotic::database {
 
 struct SnapshotHeader {
     char magic[8] = {'C', 'A', 'L', 'S', 'N', 'A', 'P', '1'};
-    uint32_t version = 3;
+    uint32_t version = 4;   // v4 = streaming/chunked body
     uint64_t timestamp = 0;
     uint64_t walSequence = 0;
     uint64_t documentCount = 0;

tests/test_snapshot_durability.cpp

@@ -276,6 +276,47 @@ void test_loadWithFallback_throws_when_all_corrupt() {
     std::cout << "PASS: loadWithFallback throws when all snapshots corrupt\n";
 }
 
+// ──────────────────────────────────────────────────────────
+// Test 7: v4 chunked format round-trips at scale (multi-chunk path)
+// ──────────────────────────────────────────────────────────
+void test_v4_format_roundtrip_at_scale() {
+    auto dir = makeSnapshotDir("v4scale");
+
+    SnapshotManager::Config scfg;
+    scfg.snapshotDir = dir;
+    scfg.compressionEnabled = true;
+    SnapshotManager mgr(scfg);
+
+    MemoryStore::Config storeCfg = defaultStoreConfig();
+    storeCfg.maxMemoryBytes = 512ULL * 1024 * 1024;  // room for 10K x ~2 KB docs
+    MemoryStore store(storeCfg);
+    store.start();
+    store.createCollection("docs", CollectionOptions{});
+    // Make each doc ~2 KB so 10,000 docs > 16 MB, forcing multi-chunk body.
+    std::string bigString(2000, 'x');
+    for (int i = 0; i < 10000; i++) {
+        Document d;
+        d.id = "d" + std::to_string(i);
+        d.data = nlohmann::json{{"value", bigString}, {"i", i}};
+        store.insert("docs", d);
+    }
+
+    auto path = mgr.createSnapshot(store, 1);
+    store.stop();
+
+    MemoryStore store2(storeCfg);
+    store2.start();
+    uint64_t seq = mgr.loadSnapshot(path, store2);
+    assert(seq == 1);
+    auto d = store2.get("docs", "d9999");
+    assert(d.has_value());
+    assert((*d).data["i"].get<int>() == 9999);
+    store2.stop();
+
+    fs::remove_all(dir);
+    std::cout << "PASS: v4 chunked format round-trips at scale (10K docs, 20MB+)\n";
+}
+
 int main() {
     test_atomic_write_leaves_no_tmp();
     test_snapshot_round_trip();
@@ -283,6 +324,7 @@ int main() {
     test_loadWithFallback_uses_older_when_newest_corrupt();
     test_orphaned_tmp_cleaned_by_listSnapshots();
     test_loadWithFallback_throws_when_all_corrupt();
+    test_v4_format_roundtrip_at_scale();
     std::cout << "\nAll snapshot durability tests PASSED!\n";
     return 0;
 }