feat: read/write data integrity check (block_hash + integrity config)

kv_cache_manager/client/include/common.h

@@ -60,6 +60,10 @@ enum [[nodiscard]] ClientErrorCode : int32_t{
     ER_CUDA_STREAM_SYNCHRONIZE_ERROR = 115,
     ER_CUDA_STREAM_DESTROY_ERROR = 116,
     ER_CUDA_HOST_REGISTER_ERROR = 117,
+
+    // data integrity
+    ER_CHECKSUM_MISMATCH = 118,     // checksum on the read buffer differs from what meta has
+    ER_INLINE_HEADER_INVALID = 119, // inline header check failed (Scheme B, reserved, disabled)
 };
 
 enum class QueryType : int {

kv_cache_manager/client/include/manager_client.h

@@ -17,41 +17,57 @@ class ManagerClient {
     static std::unique_ptr<ManagerClient> Create(const std::string &config, InitParams &init_params);
 
     // for meta client
-    virtual std::pair<ClientErrorCode, Locations>
-    MatchLocation(const std::string &trace_id,
-                  QueryType query_type,
-                  const std::vector<int64_t> &keys,
-                  const std::vector<int64_t> &tokens,
-                  const BlockMask &block_mask,
-                  int32_t sw_size,
-                  const std::vector<std::string> &location_spec_names) = 0;
+    // out_checksums (optional): when non-null, filled with the per-block checksum the
+    // server stored at write time, parallel to the returned Locations. The caller is
+    // expected to pass this same vector to LoadKvCaches's expected_checksums to enable
+    // read-side verification.
+    virtual std::pair<ClientErrorCode, Locations> MatchLocation(const std::string &trace_id,
+                                                                QueryType query_type,
+                                                                const std::vector<int64_t> &keys,
+                                                                const std::vector<int64_t> &tokens,
+                                                                const BlockMask &block_mask,
+                                                                int32_t sw_size,
+                                                                const std::vector<std::string> &location_spec_names,
+                                                                std::vector<int64_t> *out_checksums = nullptr) = 0;
 
     virtual std::pair<ClientErrorCode, WriteLocation>
     StartWrite(const std::string &trace_id,
                const std::vector<int64_t> &keys,
                const std::vector<int64_t> &tokens,
                const std::vector<std::string> &location_spec_group_names,
                int64_t write_timeout_seconds) = 0;
+    // checksums parallels the keys captured at StartWrite (length == keys.size, full
+    // batch including failed positions which carry 0). Pass an empty vector when the
+    // client did not enable meta_checksum; server then keeps existing CacheLocation
+    // checksums untouched.
     virtual ClientErrorCode FinishWrite(const std::string &trace_id,
                                         const std::string &write_session_id,
                                         const BlockMask &success_block,
-                                        const Locations &locations) = 0;
+                                        const Locations &locations,
+                                        const std::vector<int64_t> &checksums = {}) = 0;
 
     virtual std::pair<ClientErrorCode, Metas> MatchMeta(const std::string &trace_id,
                                                         const std::vector<int64_t> &keys,
                                                         const std::vector<int64_t> &tokens,
                                                         const BlockMask &block_mask,
-                                                        int32_t detail_level) = 0;
+                                                        int32_t detail_level,
+                                                        std::vector<int64_t> *out_checksums = nullptr) = 0;
 
     virtual ClientErrorCode RemoveCache(const std::string &trace_id,
                                         const std::vector<int64_t> &keys,
                                         const std::vector<int64_t> &tokens,
                                         const BlockMask &block_mask) = 0;
 
     // for transfer client
-    virtual ClientErrorCode LoadKvCaches(const UriStrVec &uri_str_vec, const BlockBuffers &block_buffers) = 0;
+    // out_checksums is forwarded to TransferClient::SaveKvCaches; when non-null the SDK
+    // writes the computed checksums into it and the caller passes the same vector to
+    // FinishWrite later.
+    virtual ClientErrorCode LoadKvCaches(const UriStrVec &uri_str_vec,
+                                         const BlockBuffers &block_buffers,
+                                         const std::vector<int64_t> *expected_checksums = nullptr) = 0;
     virtual std::pair<ClientErrorCode, UriStrVec> SaveKvCaches(const UriStrVec &uri_str_vec,
-                                                               const BlockBuffers &block_buffers) = 0;
+                                                               const BlockBuffers &block_buffers,
+                                                               std::vector<int64_t> *out_checksums = nullptr) = 0;
 
 protected:
     ManagerClient() = default;

kv_cache_manager/client/include/meta_client.h

@@ -16,31 +16,39 @@ class MetaClient {
     virtual ~MetaClient() = default;
     static std::unique_ptr<MetaClient> Create(const std::string &config, const InitParams &init_params);
 
-    virtual std::pair<ClientErrorCode, Locations>
-    MatchLocation(const std::string &trace_id,
-                  QueryType query_type,
-                  const std::vector<int64_t> &keys,
-                  const std::vector<int64_t> &tokens,
-                  const BlockMask &block_mask,
-                  int32_t sw_size,
-                  const std::vector<std::string> &location_spec_names) = 0;
+    // out_checksums (optional): filled with the per-block checksum the server stored
+    // at write time, parallel to the returned Locations. Pass nullptr to keep the
+    // legacy behavior (no verification possible on read).
+    virtual std::pair<ClientErrorCode, Locations> MatchLocation(const std::string &trace_id,
+                                                                QueryType query_type,
+                                                                const std::vector<int64_t> &keys,
+                                                                const std::vector<int64_t> &tokens,
+                                                                const BlockMask &block_mask,
+                                                                int32_t sw_size,
+                                                                const std::vector<std::string> &location_spec_names,
+                                                                std::vector<int64_t> *out_checksums = nullptr) = 0;
 
     virtual std::pair<ClientErrorCode, WriteLocation>
     StartWrite(const std::string &trace_id,
                const std::vector<int64_t> &keys,
                const std::vector<int64_t> &tokens,
                const std::vector<std::string> &location_spec_group_names,
                int64_t write_timeout_seconds) = 0;
+    // checksums parallels the keys captured at StartWrite (full batch, 0 means "not
+    // reported"). MetaClientImpl fills each one into the corresponding
+    // FinishWriteCacheRequest.locations[i].checksum field on the wire.
     virtual ClientErrorCode FinishWrite(const std::string &trace_id,
                                         const std::string &write_session_id,
                                         const BlockMask &success_block,
-                                        const Locations &locations) = 0;
+                                        const Locations &locations,
+                                        const std::vector<int64_t> &checksums = {}) = 0;
 
     virtual std::pair<ClientErrorCode, Metas> MatchMeta(const std::string &trace_id,
                                                         const std::vector<int64_t> &keys,
                                                         const std::vector<int64_t> &tokens,
                                                         const BlockMask &block_mask,
-                                                        int32_t detail_level) = 0;
+                                                        int32_t detail_level,
+                                                        std::vector<int64_t> *out_checksums = nullptr) = 0;
 
     virtual std::pair<ClientErrorCode, int64_t> MatchLocationLen(const std::string &trace_id,
                                                                  QueryType query_type,

kv_cache_manager/client/include/transfer_client.h

@@ -16,13 +16,30 @@ class TransferClient {
     virtual ~TransferClient() = default;
     static std::unique_ptr<TransferClient> Create(const std::string &client_config, const InitParams &init_params);
 
+    // Optional expected_checksums enables read-side verification. Callers (typically
+    // ManagerClient) feed the CacheLocation.checksum values returned by meta service.
+    //   - nullptr or empty vector: verification skipped (matches legacy behavior).
+    //   - size must equal block_buffers.size() (1:1).
+    //   - An element == 0 is treated as "no checksum for this block" and skipped
+    //     individually (compat with legacy data).
+    //   - Any mismatch -> ER_CHECKSUM_MISMATCH; buffer contents may be partially
+    //     written and should be discarded by the caller.
     virtual ClientErrorCode LoadKvCaches(const UriStrVec &uri_str_vec,
                                          const BlockBuffers &block_buffers,
-                                         std::shared_ptr<TransferTraceInfo> trace_info = nullptr) = 0;
-    virtual std::pair<ClientErrorCode, UriStrVec>
-    SaveKvCaches(const UriStrVec &uri_str_vec,
-                 const BlockBuffers &block_buffers,
-                 std::shared_ptr<TransferTraceInfo> trace_info = nullptr) = 0;
+                                         std::shared_ptr<TransferTraceInfo> trace_info = nullptr,
+                                         const std::vector<int64_t> *expected_checksums = nullptr) = 0;
+
+    // Optional out_checksums collects the per-block checksum the SDK computed during
+    // write, for the caller to forward to FinishWrite.
+    //   - nullptr: no checksum computed (matches legacy behavior).
+    //   - Non-null but built without CUDA/MUSA: vector cleared + warn log (caller
+    //     degrades to no checksum upstream).
+    //   - On success, vector size matches block_buffers.size(); if the upstream needs
+    //     to pad zero entries for failed blocks within a mask, that is the caller's job.
+    virtual std::pair<ClientErrorCode, UriStrVec> SaveKvCaches(const UriStrVec &uri_str_vec,
+                                                               const BlockBuffers &block_buffers,
+                                                               std::shared_ptr<TransferTraceInfo> trace_info = nullptr,
+                                                               std::vector<int64_t> *out_checksums = nullptr) = 0;
 
 protected:
     TransferClient() = default;

kv_cache_manager/client/pybind/py_client_binding.cc

@@ -124,17 +124,34 @@ PYBIND11_MODULE(kvcm_py_client, module) {
     // 绑定TransferClient类
     py::class_<kvcm::TransferClient, py::smart_holder>(module, "TransferClient")
         .def_static("Create", &kvcm::TransferClient::Create, py::call_guard<py::gil_scoped_release>())
-        .def("LoadKvCaches",
-             &kvcm::TransferClient::LoadKvCaches,
-             py::arg("uri_str_vec"),
-             py::arg("block_buffers"),
-             py::arg("trace_info") = nullptr,
-             py::call_guard<py::gil_scoped_release>())
-        .def("SaveKvCaches",
-             &kvcm::TransferClient::SaveKvCaches,
-             py::arg("uri_str_vec"),
-             py::arg("block_buffers"),
-             py::arg("trace_info") = nullptr,
-             py::call_guard<py::gil_scoped_release>());
+        // C++ interface gained pointer parameters expected_checksums / out_checksums for
+        // Scheme A verification. The Python binding intentionally keeps the legacy
+        // 3-argument signature via a lambda (default nullptr), letting connectors
+        // (vLLM / SGLang / TRT-LLM) keep building unchanged. Extend here once
+        // py_connector adopts checksum reporting.
+        .def(
+            "LoadKvCaches",
+            [](kvcm::TransferClient *self,
+               const kvcm::UriStrVec &uri_str_vec,
+               const kvcm::BlockBuffers &block_buffers,
+               std::shared_ptr<kvcm::TransferTraceInfo> trace_info) {
+                return self->LoadKvCaches(uri_str_vec, block_buffers, trace_info);
+            },
+            py::arg("uri_str_vec"),
+            py::arg("block_buffers"),
+            py::arg("trace_info") = nullptr,
+            py::call_guard<py::gil_scoped_release>())
+        .def(
+            "SaveKvCaches",
+            [](kvcm::TransferClient *self,
+               const kvcm::UriStrVec &uri_str_vec,
+               const kvcm::BlockBuffers &block_buffers,
+               std::shared_ptr<kvcm::TransferTraceInfo> trace_info) {
+                return self->SaveKvCaches(uri_str_vec, block_buffers, trace_info);
+            },
+            py::arg("uri_str_vec"),
+            py::arg("block_buffers"),
+            py::arg("trace_info") = nullptr,
+            py::call_guard<py::gil_scoped_release>());
 
 } // namespace kv_cache_manager