Refine checksum API and finish-write contract

Author: charpty

docs/design/data_integrity.md

@@ -3,6 +3,7 @@
 #include <map>
 #include <memory>
 #include <string>
+#include <utility>
 #include <variant>
 #include <vector>
 
@@ -253,36 +254,36 @@ struct MatchMetaResult {
 };
 
 struct FinishWriteOptions {
-    const std::vector<int64_t> *checksums{nullptr};
+    std::vector<int64_t> checksums;
 
-    static FinishWriteOptions WithChecksums(const std::vector<int64_t> &checksums) {
+    static FinishWriteOptions WithChecksums(std::vector<int64_t> checksums) {
         FinishWriteOptions options;
-        options.checksums = &checksums;
+        options.checksums = std::move(checksums);
         return options;
     }
 };
 
 struct LoadKvCachesOptions {
     std::shared_ptr<TransferTraceInfo> trace_info{nullptr};
-    const std::vector<int64_t> *expected_checksums{nullptr};
+    std::vector<int64_t> expected_checksums;
 
     static LoadKvCachesOptions WithTraceInfo(std::shared_ptr<TransferTraceInfo> trace_info) {
         LoadKvCachesOptions options;
         options.trace_info = trace_info;
         return options;
     }
 
-    static LoadKvCachesOptions VerifyWith(const std::vector<int64_t> &checksums) {
+    static LoadKvCachesOptions VerifyWith(std::vector<int64_t> checksums) {
         LoadKvCachesOptions options;
-        options.expected_checksums = &checksums;
+        options.expected_checksums = std::move(checksums);
         return options;
     }
 
-    static LoadKvCachesOptions VerifyWith(const std::vector<int64_t> &checksums,
+    static LoadKvCachesOptions VerifyWith(std::vector<int64_t> checksums,
                                           std::shared_ptr<TransferTraceInfo> trace_info) {
         LoadKvCachesOptions options;
         options.trace_info = trace_info;
-        options.expected_checksums = &checksums;
+        options.expected_checksums = std::move(checksums);
         return options;
     }
 };

kv_cache_manager/client/include/common.h

@@ -90,25 +90,6 @@ GenLocations(const google::protobuf::RepeatedPtrField<::kv_cache_manager::proto:
     return result;
 }
 
-kv_cache_manager::ClientErrorCode
-GenCacheLocation(const kv_cache_manager::Locations &locations,
-                 google::protobuf::RepeatedPtrField<::kv_cache_manager::proto::meta::CacheLocation> *proto_locations) {
-    if (locations.empty()) {
-        return kv_cache_manager::ClientErrorCode::ER_OK;
-    }
-    for (const auto &location : locations) {
-        auto *cache_location = proto_locations->Add();
-        cache_location->set_type(::kv_cache_manager::proto::meta::StorageType::ST_UNSPECIFIED);
-        cache_location->set_spec_size(-1);
-        for (const auto &location_spec : location) {
-            auto *spec = cache_location->add_location_specs();
-            spec->set_name(location_spec.spec_name);
-            spec->set_uri(location_spec.uri);
-        }
-    }
-    return kv_cache_manager::ClientErrorCode::ER_OK;
-}
-
 template <typename ProtoMessage>
 inline std::enable_if_t<std::is_base_of_v<google::protobuf::Message, ProtoMessage>>
 SetCommonInfo(ProtoMessage &proto_message, const std::string &trace_id, const std::string &instance_id) {
@@ -433,30 +414,8 @@ ClientErrorCode GrpcStub::FinishWriteCache(const std::string &trace_id,
     proto::meta::FinishWriteCacheRequest request;
     SetCommonInfo(request, trace_id, instance_id);
     request.set_write_session_id(write_session_id);
-    auto proto_locations = request.mutable_locations();
-    auto ec = GenCacheLocation(locations, proto_locations);
-    if (ec != ER_OK) {
-        KVCM_LOG_DEBUG("finish write cache failed, write_session_id: %s, block_mask: %s, locations: %s",
-                       write_session_id.c_str(),
-                       DebugStringUtil::ToString(success_block).c_str(),
-                       DebugStringUtil::ToString(locations).c_str());
-        return ec;
-    }
-    // Fill each checksum into the parallel CacheLocation slot. Length mismatch is a
-    // client-side bug -- fail loudly instead of dropping checksums silently, otherwise
-    // the server commits the write with checksum=0 while the caller believes the
-    // checksum was reported and an undetectable read corruption window opens.
-    if (!checksums.empty()) {
-        if (static_cast<int>(checksums.size()) != proto_locations->size()) {
-            KVCM_LOG_ERROR("checksums size [%zu] mismatches locations size [%d]; refuse to send "
-                           "FinishWriteCache without checksums",
-                           checksums.size(),
-                           proto_locations->size());
-            return ER_INVALID_PARAMS;
-        }
-        for (int i = 0; i < proto_locations->size(); ++i) {
-            proto_locations->Mutable(i)->set_checksum(checksums[i]);
-        }
+    for (auto checksum : checksums) {
+        request.add_checksums(checksum);
     }
     ProtoConvert::BlockMaskToProto(success_block, request.mutable_success_blocks());
     grpc::ClientContext context;

kv_cache_manager/client/src/internal/stub/grpc_stub.cc

@@ -475,6 +475,60 @@ TEST_F(GrpcStubTest, TestFinishWriteCacheSuccess) {
     }
 }
 
+TEST_F(GrpcStubTest, TestFinishWriteCacheWithChecksums) {
+    auto expected = std::pair<ClientErrorCode, std::string>(ER_OK, default_storage_configs);
+    ASSERT_EQ(expected,
+              stub_->RegisterInstance(
+                  "trace1", "default", "instance1", 64, createLocationSpecInfos(2), createModelDeployment(2, 1), {}));
+    std::string write_session_id;
+    Locations target_locations;
+    {
+        auto [success, write_location] = stub_->StartWriteCache("trace2", "instance1", {1, 2, 3, 4}, {}, {}, 1000000);
+        ASSERT_EQ(ER_OK, success);
+        write_session_id = write_location.write_session_id;
+        target_locations = write_location.locations;
+    }
+    {
+        BlockMask success_block = static_cast<size_t>(4);
+        Locations unrelated_locations = {{{"tp0", "file://caller-owned-location-subset"}}};
+        const std::vector<int64_t> checksums = {0x11, 0x22, 0x33, 0x44};
+        ASSERT_EQ(ER_OK,
+                  stub_->FinishWriteCache(
+                      "trace3", "instance1", write_session_id, success_block, unrelated_locations, checksums));
+    }
+    {
+        auto [success, result] = stub_->GetCacheLocation("trace4",
+                                                         "instance1",
+                                                         QueryType::QT_PREFIX_MATCH,
+                                                         {1, 2, 3, 4},
+                                                         {},
+                                                         static_cast<size_t>(0),
+                                                         {},
+                                                         MatchLocationOptions::WithChecksums());
+        ASSERT_EQ(ER_OK, success);
+        ExpectLocationsEq(target_locations, result.locations);
+        ASSERT_EQ((std::vector<int64_t>{0x11, 0x22, 0x33, 0x44}), result.checksums);
+    }
+}
+
+TEST_F(GrpcStubTest, TestFinishWriteCacheRejectsChecksumSizeMismatch) {
+    auto expected = std::pair<ClientErrorCode, std::string>(ER_OK, default_storage_configs);
+    ASSERT_EQ(expected,
+              stub_->RegisterInstance(
+                  "trace1", "default", "instance1", 64, createLocationSpecInfos(2), createModelDeployment(2, 1), {}));
+    std::string write_session_id;
+    {
+        auto [success, write_location] = stub_->StartWriteCache("trace2", "instance1", {1, 2, 3, 4}, {}, {}, 1000000);
+        ASSERT_EQ(ER_OK, success);
+        write_session_id = write_location.write_session_id;
+    }
+    {
+        BlockMask success_block = static_cast<size_t>(4);
+        ASSERT_EQ(ER_SERVICE_INVALID_ARGUMENT,
+                  stub_->FinishWriteCache("trace3", "instance1", write_session_id, success_block, {}, {0x11}));
+    }
+}
+
 TEST_F(GrpcStubTest, TestFinishWriteCacheFail) {
     auto expected = std::pair<ClientErrorCode, std::string>(ER_OK, default_storage_configs);
     ASSERT_EQ(expected,

kv_cache_manager/client/src/internal/stub/test/grpc_stub_test.cc

@@ -7,23 +7,18 @@
 namespace kv_cache_manager {
 
 // Walk the batch and report the indices of every block whose checksum disagrees.
-// The caller uses these indices to log per-block diagnostics or publish a
-// per-block ChecksumMismatchEvent.
+// The caller uses these indices to log per-block diagnostics.
 //
 // Sentinels: expected[i] == 0 means "no checksum was stored for this block"
 // (legacy data or legacy client). Such entries are skipped in both stages.
 //
-// strict_mode is kept for API compatibility with earlier revisions; verification
-// is always per-block so the function cannot accept XOR-cancelled batches.
 struct ChecksumVerifyResult {
     bool mismatch = false;
     std::vector<std::size_t> faulty_indices; // populated only when mismatch == true
 };
 
 inline ChecksumVerifyResult VerifyBatchChecksums(const std::vector<std::int64_t> &expected,
-                                                 const std::vector<std::int64_t> &actual,
-                                                 bool strict_mode) {
-    (void)strict_mode;
+                                                 const std::vector<std::int64_t> &actual) {
     ChecksumVerifyResult result;
     if (expected.size() != actual.size()) {
         result.mismatch = true;

kv_cache_manager/client/src/internal/util/checksum_verify_util.h

@@ -8,116 +8,89 @@ using namespace kv_cache_manager;
 class ChecksumVerifyUtilTest : public TESTBASE {};
 
 // 所有 block 一致 -> mismatch=false, faulty_indices 空。
-TEST_F(ChecksumVerifyUtilTest, FastPathAllMatch) {
+TEST_F(ChecksumVerifyUtilTest, AllMatch) {
     std::vector<int64_t> expected = {0x1111, 0x2222, 0x3333};
     std::vector<int64_t> actual = expected;
-    auto r = VerifyBatchChecksums(expected, actual, /*strict_mode=*/false);
+    auto r = VerifyBatchChecksums(expected, actual);
     EXPECT_FALSE(r.mismatch);
     EXPECT_TRUE(r.faulty_indices.empty());
 }
 
 // 检测到不匹配后应该回填 faulty_indices 让上层逐块打日志。
-TEST_F(ChecksumVerifyUtilTest, FastPathDetectsMismatchAndLocatesIndex) {
+TEST_F(ChecksumVerifyUtilTest, DetectsMismatchAndLocatesIndex) {
     std::vector<int64_t> expected = {0x1111, 0x2222, 0x3333};
     std::vector<int64_t> actual = {0x1111, 0xFFFF, 0x3333}; // block #1 错
-    auto r = VerifyBatchChecksums(expected, actual, /*strict_mode=*/false);
+    auto r = VerifyBatchChecksums(expected, actual);
     ASSERT_TRUE(r.mismatch);
     ASSERT_EQ(r.faulty_indices.size(), 1u);
     EXPECT_EQ(r.faulty_indices[0], 1u);
 }
 
 // 多个错位 block：把所有错的都列出来。
-TEST_F(ChecksumVerifyUtilTest, FastPathListsAllFaultyBlocks) {
+TEST_F(ChecksumVerifyUtilTest, ListsAllFaultyBlocks) {
     std::vector<int64_t> expected = {0x1111, 0x2222, 0x3333, 0x4444};
     std::vector<int64_t> actual = {0xAAAA, 0x2222, 0xBBBB, 0x4444}; // #0, #2 错
-    auto r = VerifyBatchChecksums(expected, actual, /*strict_mode=*/false);
+    auto r = VerifyBatchChecksums(expected, actual);
     ASSERT_TRUE(r.mismatch);
     ASSERT_EQ(r.faulty_indices.size(), 2u);
     EXPECT_EQ(r.faulty_indices[0], 0u);
     EXPECT_EQ(r.faulty_indices[1], 2u);
 }
 
 // expected[i] == 0 是 sentinel (legacy data / legacy client)，跳过比对。
-TEST_F(ChecksumVerifyUtilTest, FastPathSentinelZeroIsSkipped) {
+TEST_F(ChecksumVerifyUtilTest, SentinelZeroIsSkipped) {
     std::vector<int64_t> expected = {0x1111, 0, 0x3333};        // block #1 没有 checksum
     std::vector<int64_t> actual = {0x1111, 0xDEADBEEF, 0x3333}; // #1 的 actual 不会被比较
-    auto r = VerifyBatchChecksums(expected, actual, /*strict_mode=*/false);
+    auto r = VerifyBatchChecksums(expected, actual);
     EXPECT_FALSE(r.mismatch);
 }
 
 // 全 sentinel：没有可比较的项，等同于全 match。
-TEST_F(ChecksumVerifyUtilTest, FastPathAllSentinelsTreatedAsMatch) {
+TEST_F(ChecksumVerifyUtilTest, AllSentinelsTreatedAsMatch) {
     std::vector<int64_t> expected = {0, 0, 0};
     std::vector<int64_t> actual = {0xAA, 0xBB, 0xCC};
-    auto r = VerifyBatchChecksums(expected, actual, /*strict_mode=*/false);
+    auto r = VerifyBatchChecksums(expected, actual);
     EXPECT_FALSE(r.mismatch);
 }
 
 // Size mismatch (上层 bug): mismatch=true 且 faulty_indices 空 -> 上层走 size 错误日志。
 TEST_F(ChecksumVerifyUtilTest, SizeMismatchReturnsMismatchWithoutIndices) {
     std::vector<int64_t> expected = {0x1111, 0x2222};
     std::vector<int64_t> actual = {0x1111};
-    auto r = VerifyBatchChecksums(expected, actual, /*strict_mode=*/false);
+    auto r = VerifyBatchChecksums(expected, actual);
     EXPECT_TRUE(r.mismatch);
     EXPECT_TRUE(r.faulty_indices.empty());
 }
 
-// strict_mode 参数保留兼容；当前实现始终逐块比对。
-TEST_F(ChecksumVerifyUtilTest, StrictModeMatchesFastFallback) {
-    std::vector<int64_t> expected = {0x1111, 0, 0x3333, 0x4444};
-    std::vector<int64_t> actual = {0xAAAA, 0x2222, 0x3333, 0xBBBB}; // #0, #3 错；#1 是 sentinel
-    auto r = VerifyBatchChecksums(expected, actual, /*strict_mode=*/true);
-    ASSERT_TRUE(r.mismatch);
-    ASSERT_EQ(r.faulty_indices.size(), 2u);
-    EXPECT_EQ(r.faulty_indices[0], 0u);
-    EXPECT_EQ(r.faulty_indices[1], 3u);
-}
-
-// Strict mode + all match: 仍然 mismatch=false。
-TEST_F(ChecksumVerifyUtilTest, StrictModeAllMatch) {
-    std::vector<int64_t> expected = {0x1111, 0x2222, 0x3333};
-    std::vector<int64_t> actual = expected;
-    auto r = VerifyBatchChecksums(expected, actual, /*strict_mode=*/true);
-    EXPECT_FALSE(r.mismatch);
-}
-
 // Block swap (读串): expected=[A,B], actual=[B,A]. 必须识别并回填两个 faulty index。
-TEST_F(ChecksumVerifyUtilTest, FastPathCatchesBlockSwap) {
+TEST_F(ChecksumVerifyUtilTest, CatchesBlockSwap) {
     std::vector<int64_t> expected = {0xAAAA, 0xBBBB};
     std::vector<int64_t> actual = {0xBBBB, 0xAAAA};
-    auto r_fast = VerifyBatchChecksums(expected, actual, /*strict_mode=*/false);
-    ASSERT_TRUE(r_fast.mismatch);
-    ASSERT_EQ(r_fast.faulty_indices.size(), 2u);
-    EXPECT_EQ(r_fast.faulty_indices[0], 0u);
-    EXPECT_EQ(r_fast.faulty_indices[1], 1u);
-    // Strict 一致。
-    auto r_strict = VerifyBatchChecksums(expected, actual, /*strict_mode=*/true);
-    ASSERT_TRUE(r_strict.mismatch);
-    EXPECT_EQ(r_strict.faulty_indices.size(), 2u);
+    auto r = VerifyBatchChecksums(expected, actual);
+    ASSERT_TRUE(r.mismatch);
+    ASSERT_EQ(r.faulty_indices.size(), 2u);
+    EXPECT_EQ(r.faulty_indices[0], 0u);
+    EXPECT_EQ(r.faulty_indices[1], 1u);
 }
 
-// Same-delta 成对突变：每个 block 都被同一 delta 改写 (expected=[A,B],
-// actual=[A^X, B^X])。老 XOR fast 会 delta 对消而漏；逐块比对必须识别。
+// Same-delta 成对突变：每个 block 都被同一 delta 改写。
+// 逐块比对不能让这种 batch 通过。
 TEST_F(ChecksumVerifyUtilTest, DetectsSameDeltaPairedMutation) {
     constexpr int64_t kDelta = 0x0F0F0F0F0F0F0F0FLL;
     std::vector<int64_t> expected = {0xAAAA, 0xBBBB};
     std::vector<int64_t> actual = {0xAAAA ^ kDelta, 0xBBBB ^ kDelta};
-    auto r_fast = VerifyBatchChecksums(expected, actual, /*strict_mode=*/false);
-    ASSERT_TRUE(r_fast.mismatch);
-    EXPECT_EQ(r_fast.faulty_indices.size(), 2u);
-    auto r_strict = VerifyBatchChecksums(expected, actual, /*strict_mode=*/true);
-    ASSERT_TRUE(r_strict.mismatch);
-    EXPECT_EQ(r_strict.faulty_indices.size(), 2u);
+    auto r = VerifyBatchChecksums(expected, actual);
+    ASSERT_TRUE(r.mismatch);
+    EXPECT_EQ(r.faulty_indices.size(), 2u);
 }
 
-// High-bit 成对突变：奇数乘法聚合也会让最高位 delta 在偶数个 block 中抵消。
-// 逐块比对不能接受这种 batch。
+// High-bit 成对突变：逐块比对不能接受这种 batch。
 TEST_F(ChecksumVerifyUtilTest, DetectsHighBitPairedMutation) {
     constexpr int64_t kHighBit = static_cast<int64_t>(0x8000000000000000ULL);
     std::vector<int64_t> expected = {0x1111, 0x2222, 0x3333};
     std::vector<int64_t> actual = {0x1111 ^ kHighBit, 0x2222 ^ kHighBit, 0x3333};
 
-    auto r = VerifyBatchChecksums(expected, actual, /*strict_mode=*/false);
+    auto r = VerifyBatchChecksums(expected, actual);
     ASSERT_TRUE(r.mismatch);
     ASSERT_EQ(r.faulty_indices.size(), 2u);
     EXPECT_EQ(r.faulty_indices[0], 0u);

kv_cache_manager/client/src/internal/util/test/checksum_verify_util_test.cc

@@ -174,17 +174,16 @@ ClientErrorCode MetaClientImpl::FinishWrite(const std::string &trace_id,
                                             const BlockMask &success_block,
                                             const Locations &locations,
                                             const FinishWriteOptions &options) {
-    const std::vector<int64_t> empty_checksums;
-    const auto &checksums = options.checksums == nullptr ? empty_checksums : *options.checksums;
     KVCM_LOG_DEBUG("finish write with trace_id [%s], write_session_id [%s], block_mask %s, locations %s, "
                    "checksums_size %zu",
                    trace_id.c_str(),
                    write_session_id.c_str(),
                    DebugStringUtil::ToString(success_block).c_str(),
                    DebugStringUtil::ToString(locations).c_str(),
-                   checksums.size());
+                   options.checksums.size());
     const std::string &instance_id = CHECK_INSTANCE_STUB();
-    return stub_->FinishWriteCache(trace_id, instance_id, write_session_id, success_block, locations, checksums);
+    return stub_->FinishWriteCache(
+        trace_id, instance_id, write_session_id, success_block, locations, options.checksums);
 }
 
 ClientErrorCode MetaClientImpl::RemoveCache(const std::string &trace_id,