[manager/meta] optimize ReportEvent and host cache hot paths

Optimize the pure-local ReportEvent pipeline end to end: parse large HTTP bodies in request-owned buffers, use SIMD ASCII and NUL scans, flatten and intern request mutations, canonicalize versioned URIs without repeated parsing, inline the dominant one-spec task shape, and remove redundant temporary allocations.

Fuse exact single-location local RMW while preserving capacity, key-count, lifecycle, partial-failure and non-local backend semantics. Reuse LRU handles and bounded scratch, align metadata and local-LRU shard hashing, retire immutable values outside locks, and retain strict validation for every generic caller.

Accelerate GetHostCacheState with a bounded 4096-key probe and 16384-key parallel suffix, monotonic cancellation, ordinary and Mamba bitsets, request-scoped visibility parsing, serving-only filtering and strict non-NOENT error propagation. Preserve full materialization for P2P semantics and fail closed for malformed recovered EventReport URIs.

Add parser, URI, backend, capacity, lifecycle, concurrency, randomized differential, cancellation and ordered-error coverage. Document performance results, design invariants and rollback boundaries. Preserve the latest mu-main ReportEvent metrics commit unchanged.

Release validation on the merged mu-main tree passes 106 executable Bazel tests with one GPU-only skip. Pure-local measurements include about 106.32ms p50 for a 1M-key all-hit query with 4 workers, 84.38ms with 8 workers, and about 0.46ms for first-window cancellation; shared-host results are regression data, not an online SLA.

Author: charpty

docs/api/report_event.md

@@ -822,6 +822,7 @@ heartbeat/grace 短时序测试使用独立 storage/instance group，不得缩
 | D-12 | KVCM 只追加一个合法 s_version，不改变客户端 URI 其他部分 | 基础集成 `_assert_profile_specs_in_locations`；snapshot 集成 `_assert_reporter_scope` |
 | D-13 | 第一条 delta 已创建 generation、metadata 写失败时准确报错，重试复用 generation | `TestReportEventFirstDeltaMetadataFailureReportsFailureAndReusesGeneration` |
 | D-14 | 同一 spec 的折叠事件共享最终 metadata 写入失败结果 | `TestReportEventFoldedDeltaEventsShareFinalWriteFailure` |
+| D-15 | 大于 32 KiB 的部分失败批次保持逐项结果与输入索引严格对齐，只重试失败项后复用 generation 并最终收敛 | snapshot 集成 `test_34_large_partial_batch_preserves_item_alignment_and_retry` |
 | S-01 | snapshot 跨 medium 完整上报、响应返回 generation | `TestReportEventSnapshotReplacesCompleteSpecSetPerBlock`；snapshot 集成 `test_17/22` |
 | S-02 | 同 block 跨 medium 合法，同 block+medium 重复非法 | `TestReportEventRejectsCanonicalDuplicateSnapshotKeysButAllowsDifferentMedia`；snapshot 集成 `test_27_*` |
 | S-03 | snapshot block 内重复 spec name 被拒绝 | `TestReportEventRejectsDuplicateSpecNamesWithinSnapshotBlock` |

docs/design/report_event_performance.md

@@ -133,8 +133,8 @@ every `kvcm.metrics.report_interval_ms`, default 20s).
 | `manager.get_cache_location_hit_block_counter` | counter | Total blocks hit via GetCacheLocation (cumulative) |
 | `manager.prefix_match_time_us` | gauge | Outer total latency for GetHostCacheState-style prefix matching (us) |
 | `meta_searcher.indexer_get_time_us` | gauge | Wall time spent reading metadata through MetaIndexer (us) |
-| `meta_indexer.get_io_time_us` | gauge | Metadata-backend wall time; local mode includes LRU/locks/copies and does not imply Redis I/O |
-| `meta_searcher.host_projection_time_us` | gauge | GetHostCacheState visibility checks and host/spec projection time (us) |
+| `meta_indexer.get_io_time_us` | gauge | Wall-time union of metadata-backend call intervals; local mode includes LRU/locks/copies, excludes parallel projection, and does not imply Redis I/O |
+| `meta_searcher.host_projection_time_us` | gauge | Wall-time union of GetHostCacheState visibility and host/spec projection callback intervals (us) |
 | `meta_searcher.host_prefix_reduce_time_us` | gauge | GetHostCacheState normal/Mamba host-prefix reduction time (us) |
 | `meta_indexer.search_cache_hit_ratio` | gauge | Search cache hit ratio |
 | `data_storage.create_keys_counter` | counter | Total created keys |
@@ -158,8 +158,9 @@ The full list depends on the active `MetricsReporter` type. The
 The GetHostCacheState phase metrics above are nested:
 `meta_indexer.get_io_time_us` is inside `meta_searcher.indexer_get_time_us`, and
 the indexer/projection/reduction phases are inside `manager.prefix_match_time_us`.
-Do not add them together. `get_io_time_us` is a historical name; with
-`storage_type=local` it contains no Redis network operation.
+Progressive local queries pipeline backend reads with projection, so these
+intervals can overlap and must not be added together. `get_io_time_us` is a
+historical name; with `storage_type=local` it contains no Redis network operation.
 
 ## Mapping to KMonitor Metrics
 

docs/prometheus-en_US.md

@@ -130,8 +130,8 @@ kvcm_data_storage_storage_usage_ratio{type="nfs",unique_name="store_02"} 0.3
 | `manager.get_cache_location_hit_block_counter` | counter | GetCacheLocation 命中的 Block 总数（累计） |
 | `manager.prefix_match_time_us` | gauge | GetHostCacheState 等前缀匹配的外层总延迟（微秒） |
 | `meta_searcher.indexer_get_time_us` | gauge | MetaSearcher 调用 MetaIndexer 读取 metadata 的墙钟时间（微秒） |
-| `meta_indexer.get_io_time_us` | gauge | metadata backend 调用墙钟时间；local 模式也包含 LRU/锁/复制，并不表示 Redis I/O |
-| `meta_searcher.host_projection_time_us` | gauge | GetHostCacheState location 可见性检查及 host/spec 投影时间（微秒） |
+| `meta_indexer.get_io_time_us` | gauge | metadata backend 调用区间的墙钟并集；local 模式包含 LRU/锁/复制，不包含并行 projection，也不表示 Redis I/O |
+| `meta_searcher.host_projection_time_us` | gauge | GetHostCacheState location 可见性及 host/spec projection 回调区间的墙钟并集（微秒） |
 | `meta_searcher.host_prefix_reduce_time_us` | gauge | GetHostCacheState 普通/Mamba host 前缀归约时间（微秒） |
 | `meta_indexer.search_cache_hit_ratio` | gauge | 搜索缓存命中率 |
 | `data_storage.create_keys_counter` | counter | 已创建 key 总数 |
@@ -154,8 +154,8 @@ kvcm_data_storage_storage_usage_ratio{type="nfs",unique_name="store_02"} 0.3
 
 上述 GetHostCacheState 分段指标是嵌套关系：`meta_indexer.get_io_time_us` 位于
 `meta_searcher.indexer_get_time_us` 内，后者与 projection/reduce 又位于
-`manager.prefix_match_time_us` 内，排障时不能把它们相加。`get_io_time_us` 是历史命名；当实例使用
-`storage_type=local` 时没有 Redis 网络调用。
+`manager.prefix_match_time_us` 内。渐进 local 查询会流水化 backend read 与 projection，两者可能重叠，排障时
+不能把它们相加。`get_io_time_us` 是历史命名；当实例使用 `storage_type=local` 时没有 Redis 网络调用。
 
 ## 与 KMonitor 指标对照
 

docs/prometheus-zh_CN.md

@@ -3970,6 +3970,132 @@ def assert_single_invalid(body):
             baseline_version,
         )
 
+    def test_34_large_partial_batch_preserves_item_alignment_and_retry(self):
+        host = f"large-partial-{time.time_ns()}:8080"
+        base_key = 34_000_000_000 + time.time_ns() % 1_000_000_000
+        event_count = 512
+        invalid_indices = set(range(0, event_count, 17))
+
+        self.client.report_event(
+            _make_request(
+                self.instance_id,
+                host,
+                [_ev_node_register(["mem"])],
+                trace_id="t34_register",
+            )
+        )
+
+        events = []
+        raw_uris = {}
+        for index in range(event_count):
+            block_key = base_key + index
+            raw_uri = _build_event_report_uri(
+                host,
+                "mem",
+                {"size": str(index + 1), "source": f"large_{index}"},
+            )
+            raw_uris[index] = raw_uri
+            events.append(
+                _ev_block_add(
+                    block_key,
+                    "mem",
+                    _make_single_spec(
+                        "tp0",
+                        "not-a-uri" if index in invalid_indices else raw_uri,
+                    ),
+                )
+            )
+
+        payload = _make_request(
+            self.instance_id,
+            host,
+            events,
+            trace_id="t34_large_partial",
+        )
+        self.assertGreater(len(json.dumps(payload)), 32 * 1024)
+        partial = self.client.report_event(payload, check_ok=False)
+        self.assertEqual(
+            partial.get("header", {}).get("status", {}).get("code"),
+            "INVALID_ARGUMENT",
+            partial,
+        )
+        self.assertEqual(
+            partial.get("item_results"),
+            [
+                "INVALID_ARGUMENT" if index in invalid_indices else "OK"
+                for index in range(event_count)
+            ],
+        )
+        generation = partial.get("committed_snapshot_version", "")
+        self.assertEqual(len(generation), 32)
+        self.assertTrue(partial.get("snapshot_required"))
+
+        valid_samples = [1, event_count // 2 + 1, event_count - 1]
+        invalid_samples = [0, 17, max(invalid_indices)]
+        for index in valid_samples:
+            specs = _query_block_specs(
+                self.client,
+                self.instance_id,
+                base_key + index,
+                f"t34_valid_{index}",
+            )
+            self.assertEqual(len(specs), 1)
+            self.assertEqual(specs[0].get("name"), "tp0")
+            self.assertEqual(
+                _uri_identity_without_snapshot_version(specs[0]["uri"]),
+                _uri_identity_without_snapshot_version(raw_uris[index]),
+            )
+            self.assertEqual(
+                _snapshot_version_from_uri(self, specs[0]["uri"]),
+                generation,
+            )
+        for index in invalid_samples:
+            self.assertEqual(
+                _query_block_specs(
+                    self.client,
+                    self.instance_id,
+                    base_key + index,
+                    f"t34_invalid_absent_{index}",
+                ),
+                [],
+            )
+
+        retry_events = [
+            _ev_block_add(
+                base_key + index,
+                "mem",
+                _make_single_spec("tp0", raw_uris[index]),
+            )
+            for index in sorted(invalid_indices)
+        ]
+        retry = self.client.report_event(
+            _make_request(
+                self.instance_id,
+                host,
+                retry_events,
+                trace_id="t34_retry_failed_only",
+            )
+        )
+        self.assertEqual(retry.get("item_results", []), [])
+        self.assertEqual(retry.get("committed_snapshot_version"), generation)
+        self.assertFalse(retry.get("snapshot_required"))
+        for index in invalid_samples:
+            specs = _query_block_specs(
+                self.client,
+                self.instance_id,
+                base_key + index,
+                f"t34_retry_visible_{index}",
+            )
+            self.assertEqual(len(specs), 1)
+            self.assertEqual(
+                _uri_identity_without_snapshot_version(specs[0]["uri"]),
+                _uri_identity_without_snapshot_version(raw_uris[index]),
+            )
+            self.assertEqual(
+                _snapshot_version_from_uri(self, specs[0]["uri"]),
+                generation,
+            )
+
 # ---------------------------------------------------------------------------
 # Bench tests
 # ---------------------------------------------------------------------------

integration_test/meta_service/test_report_event_snapshot.py

@@ -13,6 +13,7 @@
 #include <memory>
 #include <string>
 #include <string_view>
+#include <vector>
 
 #include "kv_cache_manager/common/cache/cache.h"
 
@@ -43,6 +44,16 @@ class Cache {
     // Opaque handle to an entry stored in the cache.
     struct Handle {};
 
+    // Caller-owned workspace for allocation-free repeated batch lookup and
+    // release. The fields are intentionally generic storage owned by Cache;
+    // callers only prepare and pass the object back to the same cache.
+    struct BatchOperationScratch {
+        std::vector<uint32_t> hashes;
+        std::vector<size_t> shard_offsets;
+        std::vector<size_t> cursors;
+        std::vector<size_t> ordered_indices;
+    };
+
 public: // types hidden from Cache implementation
     // Pointer to cached object of unspecified type. (This type alias is
     // provided for clarity, not really for type checking.)
@@ -310,6 +321,21 @@ class Cache {
         }
     }
 
+    // Reserve implementation-specific batch workspace before entering a
+    // caller's critical section. The default backend needs no workspace.
+    virtual void PrepareBatchOperationScratch(size_t /*max_count*/, BatchOperationScratch * /*scratch*/) {}
+
+    // Scratch-aware variants preserve the ordinary Handle lifetime contract.
+    // Backends without a specialized implementation delegate to the existing
+    // methods, so decorators and alternate cache implementations keep their
+    // behavior unchanged.
+    virtual void LookupBatchWithScratch(const std::string_view *keys,
+                                        size_t count,
+                                        Handle **out_handles,
+                                        BatchOperationScratch * /*scratch*/) {
+        LookupBatch(keys, count, out_handles);
+    }
+
     // Convenience wrapper when secondary cache not supported
     inline Handle *BasicLookup(const std::string_view &key, Statistics *stats) {
         return Lookup(key, nullptr, nullptr, Priority::LOW, stats);
@@ -344,6 +370,10 @@ class Cache {
         }
     }
 
+    virtual void ReleaseBatchWithScratch(Handle *const *handles, size_t count, BatchOperationScratch * /*scratch*/) {
+        ReleaseBatch(handles, count);
+    }
+
     // Return the object assiciated with a handle returned by a successful
     // Lookup(). For historical reasons, this is also known at the "value"
     // associated with the key.
@@ -664,6 +694,17 @@ class CacheWrapper : public Cache {
         target_->LookupBatch(keys, count, out_handles);
     }
 
+    void PrepareBatchOperationScratch(size_t max_count, BatchOperationScratch *scratch) override {
+        target_->PrepareBatchOperationScratch(max_count, scratch);
+    }
+
+    void LookupBatchWithScratch(const std::string_view *keys,
+                                size_t count,
+                                Handle **out_handles,
+                                BatchOperationScratch *scratch) override {
+        target_->LookupBatchWithScratch(keys, count, out_handles, scratch);
+    }
+
     bool Ref(Handle *handle) override { return target_->Ref(handle); }
 
     using Cache::Release;
@@ -673,6 +714,10 @@ class CacheWrapper : public Cache {
 
     void ReleaseBatch(Handle *const *handles, size_t count) override { target_->ReleaseBatch(handles, count); }
 
+    void ReleaseBatchWithScratch(Handle *const *handles, size_t count, BatchOperationScratch *scratch) override {
+        target_->ReleaseBatchWithScratch(handles, count, scratch);
+    }
+
     ObjectPtr Value(Handle *handle) override { return target_->Value(handle); }
 
     bool Exists(const std::string_view &key) override { return target_->Exists(key); }

kv_cache_manager/common/cache/advanced_cache.h

@@ -842,14 +842,40 @@ const Cache::CacheItemHelper *LRUCache::GetCacheItemHelper(Handle *handle) const
 }
 
 void LRUCache::LookupBatch(const std::string_view *keys, size_t count, Handle **out_handles) {
+    BatchOperationScratch scratch;
+    PrepareBatchOperationScratch(count, &scratch);
+    LookupBatchWithScratch(keys, count, out_handles, &scratch);
+}
+
+void LRUCache::PrepareBatchOperationScratch(size_t max_count, BatchOperationScratch *scratch) {
+    if (!scratch) {
+        return;
+    }
+    scratch->hashes.reserve(max_count);
+    scratch->ordered_indices.reserve(max_count);
+    const size_t shard_slots = GetNumShards() + 1;
+    scratch->shard_offsets.reserve(shard_slots);
+    scratch->cursors.reserve(shard_slots);
+}
+
+void LRUCache::LookupBatchWithScratch(const std::string_view *keys,
+                                      size_t count,
+                                      Handle **out_handles,
+                                      BatchOperationScratch *scratch) {
     if (count == 0) {
         return;
     }
+    if (!scratch) {
+        LookupBatch(keys, count, out_handles);
+        return;
+    }
     std::fill(out_handles, out_handles + count, nullptr);
 
     const size_t shard_count = GetNumShards();
-    std::vector<uint32_t> hashes(count);
-    std::vector<size_t> shard_offsets(shard_count + 1, 0);
+    scratch->hashes.resize(count);
+    scratch->shard_offsets.assign(shard_count + 1, 0);
+    auto &hashes = scratch->hashes;
+    auto &shard_offsets = scratch->shard_offsets;
     for (size_t i = 0; i < count; ++i) {
         hashes[i] = LRUCacheShard::ComputeHash(keys[i], hash_seed_);
         const size_t shard = LRUCacheShard::HashPieceForSharding(hashes[i]) & shard_mask_;
@@ -859,8 +885,10 @@ void LRUCache::LookupBatch(const std::string_view *keys, size_t count, Handle **
         shard_offsets[shard + 1] += shard_offsets[shard];
     }
 
-    std::vector<size_t> cursors = shard_offsets;
-    std::vector<size_t> ordered_indices(count);
+    scratch->cursors.assign(shard_offsets.begin(), shard_offsets.end());
+    scratch->ordered_indices.resize(count);
+    auto &cursors = scratch->cursors;
+    auto &ordered_indices = scratch->ordered_indices;
     for (size_t i = 0; i < count; ++i) {
         const size_t shard = LRUCacheShard::HashPieceForSharding(hashes[i]) & shard_mask_;
         ordered_indices[cursors[shard]++] = i;
@@ -877,12 +905,23 @@ void LRUCache::LookupBatch(const std::string_view *keys, size_t count, Handle **
 }
 
 void LRUCache::ReleaseBatch(Handle *const *handles, size_t count) {
+    BatchOperationScratch scratch;
+    PrepareBatchOperationScratch(count, &scratch);
+    ReleaseBatchWithScratch(handles, count, &scratch);
+}
+
+void LRUCache::ReleaseBatchWithScratch(Handle *const *handles, size_t count, BatchOperationScratch *scratch) {
     if (count == 0) {
         return;
     }
+    if (!scratch) {
+        ReleaseBatch(handles, count);
+        return;
+    }
 
     const size_t shard_count = GetNumShards();
-    std::vector<size_t> shard_offsets(shard_count + 1, 0);
+    scratch->shard_offsets.assign(shard_count + 1, 0);
+    auto &shard_offsets = scratch->shard_offsets;
     size_t non_null_count = 0;
     for (size_t i = 0; i < count; ++i) {
         if (handles[i] == nullptr) {
@@ -897,8 +936,10 @@ void LRUCache::ReleaseBatch(Handle *const *handles, size_t count) {
         shard_offsets[shard + 1] += shard_offsets[shard];
     }
 
-    std::vector<size_t> cursors = shard_offsets;
-    std::vector<size_t> ordered_indices(non_null_count);
+    scratch->cursors.assign(shard_offsets.begin(), shard_offsets.end());
+    scratch->ordered_indices.resize(non_null_count);
+    auto &cursors = scratch->cursors;
+    auto &ordered_indices = scratch->ordered_indices;
     for (size_t i = 0; i < count; ++i) {
         if (handles[i] == nullptr) {
             continue;

kv_cache_manager/common/cache/lru_cache.cc

@@ -511,7 +511,13 @@ class LRUCache
     const CacheItemHelper *GetCacheItemHelper(Handle *handle) const override;
 
     void LookupBatch(const std::string_view *keys, size_t count, Handle **out_handles) override;
+    void PrepareBatchOperationScratch(size_t max_count, BatchOperationScratch *scratch) override;
+    void LookupBatchWithScratch(const std::string_view *keys,
+                                size_t count,
+                                Handle **out_handles,
+                                BatchOperationScratch *scratch) override;
     void ReleaseBatch(Handle *const *handles, size_t count) override;
+    void ReleaseBatchWithScratch(Handle *const *handles, size_t count, BatchOperationScratch *scratch) override;
 
     void ApplyToHandle(Cache *cache,
                        Handle *handle,

kv_cache_manager/common/cache/lru_cache.h

@@ -161,7 +161,13 @@ ErrorCode EventReportBackend::Close() {
     retired_.store(true, std::memory_order_release);
     SetOpen(false);
     SetAvailable(false);
-    liveness_checker_running_.store(false, std::memory_order_release);
+    // Serialize the predicate update with wait_for().  An atomic predicate is
+    // not sufficient to prevent a lost notification when the waiter is
+    // between checking the predicate and actually sleeping.
+    {
+        std::lock_guard<std::mutex> wait_guard(liveness_wait_mutex_);
+        liveness_checker_running_.store(false, std::memory_order_release);
+    }
     liveness_wait_cv_.notify_all();
     if (liveness_checker_thread_.joinable()) {
         liveness_checker_thread_.join();