[Store] Extract master snapshot codec from MasterService（3/5）

mooncake-store/include/ha/snapshot/master_snapshot_codec.h

@@ -0,0 +1,153 @@
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <vector>
+#include <ylt/util/tl/expected.hpp>
+
+#include "types.h"
+
+namespace mooncake {
+
+// Forward declarations
+class MasterService;
+class SegmentManager;
+class NoFSegmentManager;
+class ClientTaskManager;
+
+namespace ha {
+
+/**
+ * @brief A view of the live master state for snapshot serialization.
+ *
+ * This struct holds references to the live state components that need to
+ * be serialized into a master snapshot. Non-const references are used because
+ * the underlying serializers require non-const pointers, avoiding const_cast.
+ */
+struct MasterSnapshotStateView {
+    MasterService& master_service;
+    SegmentManager& segment_manager;
+    NoFSegmentManager& nof_segment_manager;
+    ClientTaskManager& task_manager;
+
+    MasterSnapshotStateView(MasterService& ms, SegmentManager& sm,
+                            NoFSegmentManager& nsm, ClientTaskManager& tm)
+        : master_service(ms),
+          segment_manager(sm),
+          nof_segment_manager(nsm),
+          task_manager(tm) {}
+};
+
+/**
+ * @brief Container for serialized master snapshot payloads.
+ *
+ * This struct provides a type-safe, zero-overhead container for the three
+ * serialized payload buffers, eliminating map lookup overhead and potential
+ * runtime exceptions from using std::unordered_map.
+ */
+struct MasterSnapshotPayloads {
+    std::vector<uint8_t> metadata;
+    std::vector<uint8_t> segments;
+    std::vector<uint8_t> task_manager;
+};
+
+/**
+ * @brief Encodes and decodes master snapshot payloads.
+ *
+ * This codec handles serialization of the complete master state bundle:
+ * - Metadata shards (objects, replicas, tenant state)
+ * - Segment manager state
+ * - Task manager state
+ * - Discarded replicas
+ *
+ * The current implementation preserves the existing snapshot format exactly
+ * to maintain backward compatibility with existing snapshots.
+ *
+ * Format details:
+ * - metadata: msgpack-encoded metadata shards (compressed per-shard with zstd)
+ * - segments: msgpack-encoded segment manager state
+ * - task_manager: msgpack-encoded task manager state
+ * - manifest.txt: format descriptor "<type>|<version>|<snapshot_id>"
+ *   (e.g., "messagepack|1.0.0|snapshot-000123")
+ */
+class MasterSnapshotCodec {
+   public:
+    MasterSnapshotCodec() = default;
+    ~MasterSnapshotCodec() = default;
+
+    // Non-copyable, non-movable (contains no state, but enforce ownership
+    // semantics)
+    MasterSnapshotCodec(const MasterSnapshotCodec&) = delete;
+    MasterSnapshotCodec& operator=(const MasterSnapshotCodec&) = delete;
+    MasterSnapshotCodec(MasterSnapshotCodec&&) = delete;
+    MasterSnapshotCodec& operator=(MasterSnapshotCodec&&) = delete;
+
+    /**
+     * @brief Encode master state into serialized buffers.
+     *
+     * @param state_view View of the live master state
+     * @return Structured payloads containing serialized data, or error
+     *
+     * The returned struct contains:
+     * - metadata: serialized metadata shards
+     * - segments: serialized segment manager state
+     * - task_manager: serialized task manager state
+     */
+    tl::expected<MasterSnapshotPayloads, SerializationError> Encode(
+        MasterSnapshotStateView& state_view) const;
+
+    /**
+     * @brief Decode snapshot payloads and restore into master service.
+     *
+     * @param master_service Target MasterService to restore state into
+     * @param payloads Structured payloads containing serialized data
+     * @return void on success, SerializationError on failure
+     */
+    tl::expected<void, SerializationError> Decode(
+        MasterService* master_service,
+        const MasterSnapshotPayloads& payloads) const;
+
+    // Canonical serializer identifiers embedded in the snapshot manifest.
+    static constexpr const char* kSerializerType = "messagepack";
+    static constexpr const char* kSerializerVersion = "1.0.0";
+
+    /**
+     * @brief Encode a snapshot manifest into its on-disk byte representation.
+     *
+     * The manifest is a "<type>|<version>|<snapshot_id>" descriptor. Keeping
+     * the encoding here (rather than hand-crafting the format string at the
+     * call site) ensures the manifest layout stays owned by the codec.
+     *
+     * @param type Serializer/protocol type (e.g., "messagepack")
+     * @param version Snapshot format version (e.g., "1.0.0")
+     * @param snapshot_id Identifier of the snapshot being written
+     * @return Manifest bytes ready to upload
+     */
+    static std::vector<uint8_t> EncodeManifest(const std::string& type,
+                                               const std::string& version,
+                                               const std::string& snapshot_id);
+
+   private:
+    // Metadata encoding/decoding (delegates to MetadataSerializer for now)
+    tl::expected<std::vector<uint8_t>, SerializationError> EncodeMetadata(
+        MasterService& master_service) const;
+    tl::expected<void, SerializationError> DecodeMetadata(
+        MasterService* master_service, const std::vector<uint8_t>& data) const;
+
+    // Segment encoding/decoding
+    tl::expected<std::vector<uint8_t>, SerializationError> EncodeSegments(
+        SegmentManager& segment_manager,
+        NoFSegmentManager& nof_segment_manager) const;
+    tl::expected<void, SerializationError> DecodeSegments(
+        MasterService* master_service, const std::vector<uint8_t>& data) const;
+
+    // Task manager encoding/decoding
+    tl::expected<std::vector<uint8_t>, SerializationError> EncodeTaskManager(
+        ClientTaskManager& task_manager) const;
+    tl::expected<void, SerializationError> DecodeTaskManager(
+        MasterService* master_service, const std::vector<uint8_t>& data) const;
+};
+
+}  // namespace ha
+}  // namespace mooncake

mooncake-store/include/ha/snapshot/snapshot_constants.h

@@ -0,0 +1,25 @@
+#pragma once
+
+namespace mooncake::ha {
+
+// Snapshot file names
+inline constexpr const char* kSnapshotMetadataFile = "metadata";
+inline constexpr const char* kSnapshotSegmentsFile = "segments";
+inline constexpr const char* kSnapshotTaskManagerFile = "task_manager";
+inline constexpr const char* kSnapshotManifestFile = "manifest.txt";
+inline constexpr const char* kSnapshotLatestFile = "latest.txt";
+
+// Snapshot format
+inline constexpr const char* kSnapshotSerializerType = "messagepack";
+inline constexpr const char* kSnapshotSerializerVersion = "1.0.0";
+
+// Backup directories
+inline constexpr const char* kSnapshotBackupSaveDir =
+    "mooncake_snapshot_save_backup";
+inline constexpr const char* kSnapshotBackupRestoreDir =
+    "mooncake_snapshot_restore_backup";
+
+// List limit
+inline constexpr size_t kUnlimitedSnapshotList = 0;
+
+}  // namespace mooncake::ha

mooncake-store/include/master_service.h

@@ -42,10 +42,14 @@ namespace mooncake {
 
 // Forward declaration for MasterSnapshotManager
 class MasterSnapshotManager;
+class MasterSnapshotRepository;
 
 namespace ha {
 class SnapshotCatalogStore;
-}
+class MasterSnapshotCodec;
+struct MasterSnapshotPayloads;
+class MasterSnapshotCodecTest;  // test fixture, needs private state access
+}  // namespace ha
 
 class EtcdOpLogStore;
 
@@ -92,8 +96,11 @@ class MasterService {
     friend class test::PromotionOnHitTest;
     friend class benchmarks::BatchEvictBench;
     friend class test::MasterServiceTenantQuotaTest;
-    friend class MasterSnapshotManager;  // Allow access to internal state for
-                                         // snapshot
+    friend class MasterSnapshotManager;    // Allow access to internal state for
+                                           // snapshot
+    friend class ha::MasterSnapshotCodec;  // Allow codec to access private
+                                           // members
+    friend class ha::MasterSnapshotCodecTest;  // codec round-trip unit test
 
    public:
     using NoFProbeFn =
@@ -811,6 +818,16 @@ class MasterService {
         const std::chrono::system_clock::time_point& now);
     void ResetStateAfterFailedRestoreAttempt();
 
+    /**
+     * @brief Apply decoded snapshot state to running master service
+     * @param payloads Decoded snapshot payloads
+     * @param now Current time for cleanup logic
+     * @return void on success, SerializationError on failure
+     */
+    tl::expected<void, SerializationError> ApplySnapshotState(
+        const ha::MasterSnapshotPayloads& payloads,
+        const std::chrono::system_clock::time_point& now);
+
     // BatchEvict evicts objects in a near-LRU way, i.e., prioritizes to evict
     // object with smaller lease timeout. It has two passes. The first pass only
     // evicts objects without soft pin. The second pass prioritizes objects
@@ -2001,6 +2018,8 @@ class MasterService {
     std::string snapshot_catalog_store_connstring_;
     std::unique_ptr<SnapshotObjectStore> snapshot_object_store_;
     std::unique_ptr<ha::SnapshotCatalogStore> snapshot_catalog_store_;
+    std::unique_ptr<MasterSnapshotRepository> snapshot_repository_;
+    std::unique_ptr<ha::MasterSnapshotCodec> snapshot_codec_;
     mutable std::shared_mutex snapshot_mutex_;
 
     // Discarded replicas management

mooncake-store/include/master_snapshot_manager.h

@@ -12,6 +12,7 @@
 
 #include "types.h"
 #include "ha/ha_types.h"
+#include "ha/snapshot/master_snapshot_codec.h"
 
 namespace mooncake {
 

mooncake-store/include/master_snapshot_repository.h

@@ -1,12 +1,14 @@
 #pragma once
 
+#include <optional>
 #include <string>
 #include <vector>
 
 #include <ylt/util/tl/expected.hpp>
 
 #include "types.h"
 #include "ha/ha_types.h"
+#include "ha/snapshot/master_snapshot_codec.h"
 
 namespace mooncake {
 
@@ -82,6 +84,28 @@ class MasterSnapshotRepository {
      */
     std::string GetObjectStoreConnectionInfo() const;
 
+    /**
+     * @brief Load the latest snapshot descriptor from catalog
+     * @return Snapshot descriptor on success, error code on failure
+     */
+    tl::expected<ha::SnapshotDescriptor, ErrorCode> LoadLatestSnapshot();
+
+    /**
+     * @brief Load all restorable snapshot descriptors
+     * @param latest_id Optional latest snapshot ID to filter candidates
+     * @return Vector of candidate snapshots in chronological order, or error
+     */
+    tl::expected<std::vector<ha::SnapshotDescriptor>, ErrorCode>
+    LoadRestoreCandidates(const std::optional<std::string>& latest_id);
+
+    /**
+     * @brief Download snapshot payloads from object storage
+     * @param descriptor Snapshot descriptor with object paths
+     * @return Structured payloads ready for decoding, or error
+     */
+    tl::expected<ha::MasterSnapshotPayloads, SerializationError>
+    DownloadSnapshotPayloads(const ha::SnapshotDescriptor& descriptor);
+
    private:
     SnapshotObjectStore* object_store_;
     ha::SnapshotCatalogStore* catalog_store_;

mooncake-store/src/CMakeLists.txt

@@ -48,6 +48,7 @@ set(MOONCAKE_STORE_SOURCES
     ha/leadership/master_service_supervisor.cpp
     ha/standby_controller.cpp
     ha/snapshot/catalog_backed_snapshot_provider.cpp
+    ha/snapshot/master_snapshot_codec.cpp
     ha/snapshot/object/snapshot_object_store.cpp
     ha/snapshot/object/backends/local/local_file_snapshot_object_store.cpp
     ha/snapshot/object/backends/s3/s3_snapshot_object_store.cpp