fix(snapshot): atomic writer + post-write verification — fixes silent data loss (Zoe)

The Zoe production instance was silently corrupted when a snapshot was
truncated mid-write: header claimed compressedSize=N bytes but only <N
were on disk. On next restart the loader failed, the service ignored
the failure, and the DB started empty — accepting writes onto a blank
store. All production data was silently overwritten.

Root causes (all in snapshot.cpp):
- Writer never checked ofstream failbit after write() calls
- Writer never fsync()'d before close()
- Writer wrote directly to the final filename — partial writes survived
- verifySnapshot() existed but was never called after write
- cleanupOldSnapshots() ran blindly, evicting good snapshots over bad ones

Fixes:
- Write to <path>.tmp first
- Check failbit after every write() — explicit error on short write
- flush() + fsync() the file before considering it durable
- std::filesystem::rename(tmp, final) — atomic on same filesystem
- fsync() the containing directory so the rename survives crash
- Call verifySnapshot() after rename; delete + throw if it fails
- cleanupOldSnapshots() only runs after verification passes
- listSnapshots() removes orphaned .tmp files from prior crashed writes

After this commit, every snapshot file in the dir is either complete
and verified, or absent. No more silent corruption.

service/src/persistence/snapshot.cpp

@@ -2,11 +2,18 @@
 #include "wal.hpp"
 
 #include <algorithm>
+#include <cerrno>
 #include <chrono>
 #include <cstring>
 #include <fstream>
 #include <iomanip>
 #include <sstream>
+#include <system_error>
+
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <spdlog/spdlog.h>
 
 #ifndef DISABLE_LZ4
 #include <lz4.h>
@@ -59,29 +66,110 @@ std::filesystem::path SnapshotManager::createSnapshot(const MemoryStore& store,
     // Calculate header checksum (excluding the checksum field itself)
     header.headerChecksum = crc32::calculate(&header, sizeof(header) - sizeof(header.headerChecksum));
 
-    // Generate filename and write
+    // Generate filename — write to .tmp first, atomic rename at the end
     std::filesystem::path snapshotPath = config_.snapshotDir / generateFilename();
+    std::filesystem::path tmpPath = snapshotPath;
+    tmpPath += ".tmp";
 
-    std::ofstream file(snapshotPath, std::ios::binary);
-    if (!file) {
-        throw std::runtime_error("Failed to create snapshot file: " + snapshotPath.string());
-    }
+    // Helper to remove tmp on error paths
+    auto cleanupTmp = [&tmpPath]() {
+        std::error_code ec;
+        std::filesystem::remove(tmpPath, ec);
+    };
+
+    // Write to temp file
+    {
+        std::ofstream file(tmpPath, std::ios::binary | std::ios::trunc);
+        if (!file) {
+            throw std::runtime_error("Failed to create snapshot tmp file: " + tmpPath.string());
+        }
+
+        // Header + short-write check
+        file.write(reinterpret_cast<const char*>(&header), sizeof(header));
+        if (!file) {
+            cleanupTmp();
+            throw std::runtime_error("Failed to write snapshot header: " + tmpPath.string());
+        }
+
+        // Body + short-write check
+        file.write(reinterpret_cast<const char*>(bodyData.data()),
+                   static_cast<std::streamsize>(bodyData.size()));
+        if (!file) {
+            cleanupTmp();
+            throw std::runtime_error("Failed to write snapshot body (short write): " + tmpPath.string());
+        }
+
+        // Trailer (body CRC) + short-write check
+        uint32_t bodyChecksum = crc32::calculate(bodyData.data(), bodyData.size());
+        file.write(reinterpret_cast<const char*>(&bodyChecksum), sizeof(bodyChecksum));
+        if (!file) {
+            cleanupTmp();
+            throw std::runtime_error("Failed to write snapshot trailer: " + tmpPath.string());
+        }
 
-    // Write header
-    file.write(reinterpret_cast<const char*>(&header), sizeof(header));
+        // Flush C++ stream buffer
+        file.flush();
+        if (!file) {
+            cleanupTmp();
+            throw std::runtime_error("Failed to flush snapshot: " + tmpPath.string());
+        }
+    }  // ofstream destructor runs close()
+
+    // fsync the file descriptor — C++ ofstream does not do this
+    {
+        int fd = ::open(tmpPath.c_str(), O_RDONLY);
+        if (fd < 0) {
+            cleanupTmp();
+            throw std::runtime_error("Failed to reopen snapshot tmp for fsync: " + tmpPath.string());
+        }
+        if (::fsync(fd) != 0) {
+            int saved_errno = errno;
+            ::close(fd);
+            cleanupTmp();
+            throw std::runtime_error("fsync failed on snapshot tmp (errno=" +
+                                     std::to_string(saved_errno) + "): " + tmpPath.string());
+        }
+        ::close(fd);
+    }
 
-    // Write body
-    file.write(reinterpret_cast<const char*>(bodyData.data()),
-               static_cast<std::streamsize>(bodyData.size()));
+    // Atomic rename — from this point on, either the final file exists and is complete,
+    // or the .tmp is gone and no snapshot was produced.
+    {
+        std::error_code ec;
+        std::filesystem::rename(tmpPath, snapshotPath, ec);
+        if (ec) {
+            cleanupTmp();
+            throw std::runtime_error("Snapshot rename failed: " + ec.message());
+        }
+    }
 
-    // Write body checksum
-    uint32_t bodyChecksum = crc32::calculate(bodyData.data(), bodyData.size());
-    file.write(reinterpret_cast<const char*>(&bodyChecksum), sizeof(bodyChecksum));
+    // fsync the containing directory so the rename is durable after crash
+    {
+        int dfd = ::open(config_.snapshotDir.c_str(), O_RDONLY | O_DIRECTORY);
+        if (dfd >= 0) {
+            ::fsync(dfd);
+            ::close(dfd);
+        }
+        // Non-fatal on failure — some filesystems reject O_DIRECTORY
+    }
 
-    file.close();
+    // Post-write verification (opt-out via config)
+    bool verified = true;
+    if (config_.validateAfterWrite) {
+        if (!verifySnapshot(snapshotPath)) {
+            // The file exists on disk but is corrupt — delete it so it can't be used
+            std::error_code ec;
+            std::filesystem::remove(snapshotPath, ec);
+            throw std::runtime_error("Snapshot verification failed after write: " + snapshotPath.string());
+        }
+        spdlog::debug("Snapshot verified after write: {}", snapshotPath.string());
+    }
 
-    // Cleanup old snapshots
-    cleanupOldSnapshots();
+    // Cleanup — only evict old snapshots if the new one passed verification
+    // (or if cleanup_only_if_verified is disabled)
+    if (verified || !config_.cleanupOnlyIfVerified) {
+        cleanupOldSnapshots();
+    }
 
     return snapshotPath;
 }
@@ -156,6 +244,21 @@ uint64_t SnapshotManager::loadSnapshot(const std::filesystem::path& path, Memory
 std::vector<std::filesystem::path> SnapshotManager::listSnapshots() const {
     std::vector<std::pair<uint64_t, std::filesystem::path>> snapshots;
 
+    // Clean up .tmp files left from crashed writes (they start with "snapshot-" and end with ".tmp")
+    {
+        std::error_code scan_ec;
+        for (const auto& entry : std::filesystem::directory_iterator(config_.snapshotDir, scan_ec)) {
+            if (!entry.is_regular_file()) continue;
+            const auto& p = entry.path();
+            if (p.extension() == ".tmp" &&
+                p.filename().string().starts_with("snapshot-")) {
+                std::error_code rm_ec;
+                std::filesystem::remove(p, rm_ec);
+                spdlog::info("Removed orphaned snapshot tmp file: {}", p.string());
+            }
+        }
+    }
+
     std::error_code ec;
     for (const auto& entry : std::filesystem::directory_iterator(config_.snapshotDir, ec)) {
         if (entry.is_regular_file()) {

service/src/persistence/snapshot.hpp

@@ -79,6 +79,10 @@ public:
         std::filesystem::path snapshotDir;
         bool compressionEnabled = true;
         uint32_t maxSnapshots = 5;  // Keep last N snapshots
+
+        // NEW in v1.6.1: durability
+        bool validateAfterWrite = true;
+        bool cleanupOnlyIfVerified = true;
     };
 
     explicit SnapshotManager(Config config);