[manager] bound P2P host cache queries and preserve APIs

Preserve the existing HostCacheMatch and location-spec contracts by adding an explicit per-key filter. Replace request-wide P2P map/set graphs with ordered bounded reducers, keep non-progressive backends at one read, harden metrics and availability fallbacks, and add directed and million-key benchmark coverage.

Author: charpty

docs/api/report_event.md

@@ -619,7 +619,9 @@ HTTP 接口为 `POST /api/getCacheLocation`：
 `ST_EVENT_REPORT_L1P5`、`ST_EVENT_REPORT_L2` 等 backend，适合验证两种 EventReport storage 的
 隔离状态。
 
-`location_spec_names` 不只是返回结果的投影条件，也是 backend/peer 选择前按 query key 生效的候选条件：
+`location_spec_names` 保持原有语义：它是应用到每个 query key 的全局返回投影；为空时返回
+选中 location 的全部 spec。需要在 backend/peer 选择前按 query key 限制候选时，使用新增的
+`per_key_location_spec_names`：
 
 - 为空时，location 中任意合法 spec 都可使该 location 成为候选；
 - 非空时，数组长度必须等于 query key 数量，且每个 name 都不能为空；第 i 个 name 只过滤第 i 个 key；
@@ -631,7 +633,7 @@ HTTP 接口为 `POST /api/getCacheLocation`：
 
 因此 spec name 是 reporter 与查询方之间的稳定协议字段，不能用 object size 代替：不同 cache
 group 即使 byte size 相同，也必须使用不同且稳定的 spec name。调用方必须让
-`location_spec_names` 与 `block_keys`（或由 token 生成的 query keys）同序对齐。同一个 block key
+`per_key_location_spec_names` 与 `block_keys`（或由 token 生成的 query keys）同序对齐。同一个 block key
 可以在不同位置重复并请求不同 spec，用于 mixed-attention/Mamba groups。长度不匹配或包含空
 name 会返回 `INVALID_ARGUMENT`。确定性 tie-break 只消除无序遍历造成的抖动；各 key 经过
 spec 过滤后的候选 peer 集合仍可能不同。
@@ -662,7 +664,7 @@ HTTP 接口为 `POST /api/getHostCacheState`：
   "hosts": [
     {
       "host_ip_port": "10.0.0.8:8080",
-      "local": "3",
+      "prefix_match_blocks": "3",
       "p2p_1_fetch": "0",
       "p2p_1_total_match": "3"
     }
@@ -680,17 +682,17 @@ HTTP 接口为 `POST /api/getHostCacheState`：
 - `QT_UNSPECIFIED` 使用 RegisterInstance 时配置的 `default_query_type`；
 - 支持 `QT_PREFIX_MATCH` 和 `QT_PREFIX_MATCH_WITH_MAMBA`，其他类型返回参数错误；
 - 同一个 host 在多个 backend 的有效 cache 会按 host 汇总参与匹配；
-- `local` 包含同一 host 的 subscriber 与 Vineyard 上报；
-- 非混合注意力对 full local-miss 使用 Prefix 选择远端 Vineyard；混合注意力先对
+- `prefix_match_blocks` 包含同一 host 的 subscriber 与 Vineyard 上报；
+- 非混合注意力对 full local miss 使用 Prefix 选择远端 Vineyard；混合注意力先对
   FullAttention group 使用 Prefix，再对 Mamba local-miss spec 使用 Coverage；
 - `p2p_1_fetch` 表示各 P2P 阶段实际选中并拉取的 spec 所属的去重 block key 数；
 - `p2p_1_total_match` 表示本地 cache 与实际选中的远端 spec 合并后的最终前缀；
-- 远端 P2P 候选只使用 `ST_EVENT_REPORT_L2`，且不会让 `local` 为 0 的 host 出现在响应中；
+- 远端 P2P 候选只使用 `ST_EVENT_REPORT_L2`，且不会让 `prefix_match_blocks` 为 0 的 host 出现在响应中；
 - reporter unavailable 时，该 host 对应的 event-report location 不参与匹配。
 
 成功完整 snapshot 后，`GetHostCacheState` 会立即忽略完全属于旧 generation 的 location。
 snapshot 失败、KVCM 重启恢复或 realtime-only reporter 仍使用 soft metadata，因此
-`local` 在这些模式下仍可能是 false positive。
+`prefix_match_blocks` 在这些模式下仍可能是 false positive。
 
 ## 12. 节点生命周期与查询
 

docs/design/report_event_performance.md

@@ -98,8 +98,9 @@ instance group、data-storage backend 和 reporter node lock。当前实现做
 10. prefix 只把首个 `EC_NOENT` 当作正常终止。首个 miss 之后的 speculative read 结果不影响已经确定的
     前缀；但 miss 之前的 `EC_ERROR`、`EC_MISMATCH` 等硬错误必须原样返回，不能伪装成较短的 cache miss。
     普通 prefix 与 Mamba 路径遵循相同规则；
-11. 上述渐进取消只用于 `p2p_host_count=0`。启用 P2P 时 local miss 之后仍可能由 Vineyard peer 延续覆盖，
-    因此当前保留完整 metadata materialization；它仍执行 serving 过滤和错误传播，但不能套用 local-only stop。
+11. `p2p_host_count>0` 不能把 local miss 直接当作全局 stop，因为 Vineyard peer 仍可能延续覆盖。该路径改用
+    ordered compact visitor：普通 prefix 在线维护候选 peer 的前缀交集；Mamba 分 local、peer plan、最终投影三个
+    有界 pass。非 local backend 只读取一次 compact batch 并在内存中重放，均不再构造逐 key 的 `map/set` 图。
 
 可见性快照在首个有界 metadata range 读取之后、其 projection 开始时采集。采集前已经可见的 HOST_DOWN
 会被当前请求过滤；采集后的 HOST_DOWN 允许当前请求继续看到旧状态，但采集完成后本请求不再变化，下一
@@ -1704,9 +1705,10 @@ pure-local Release 验证结果：
    批次会在百万 key 请求中重复约 24 万次 lookup/release shard lock；增大后续批次可摊薄 lock/unlock，同时首窗口
    仍保持早停上界。16384 相对 4096 的同机 A/B 将 1M metadata/all-hit p50 从约 109.3/114.9ms 降至
    94.3/103.7ms；32768 没有进一步改善 all-hit 且放大后缀过读，已撤销。
-3. `p2p_host_count>0` 暂时保留完整读取。一个 host 的 local prefix 已停止时，Vineyard peer 仍可能继续覆盖后续 key，
-   不能把 local miss 直接当作全局取消条件。该路径仍获得 serving 状态过滤、一次 host/spec 扫描和严格错误传播，
-   但没有伪装成渐进路径；后续若优化，必须先设计 peer-aware stop 证明。
+3. `p2p_host_count>0` 使用 peer-aware ordered reduction。普通 prefix 只为最终 top-N host 保留 peer 交集和精确
+   fetched-key 计数；Mamba 用三个有界 pass 分别求 local 状态、各 full group 的 prefix peer 与 state coverage peer、
+   最终合并结果。local backend 的常驻投影状态与 key 数无关（协议要求的去重 fetched key 向量除外）；不支持
+   progressive read 的 backend 保留一次 compact batch，后续 pass 只重放该 batch，避免重复远端 I/O。
 4. request-specific checker 除返回可见/不可见外，同时借用返回已经解析的 reporter medium/host，并声明 EventReport URI
    已通过 generation fence 校验。projection 因而不再对同一 location 重复拆 location id、解析 URI 和提取 host；
    medium filter 使用借用的 `string_view` hash set，generic URI path 也不再复制 path string。
@@ -1753,8 +1755,8 @@ location 指针带出 item lock 虽可少一次 owner add-ref，但 ReportEvent
 - serving/WRITING/DELETING/NEW/NOT_FOUND 全状态矩阵在 p2p=0 和 p2p>0 下结果一致；Manager 端到端验证
   StartWrite 在 Finish 前不可见；
 - ordinary/Mamba 覆盖 first-window stop、第二并行 range timeout、较早 visitor stop 屏蔽无关后续 timeout、70 host
-  多 presence word、75 个 required spec 跨 word，以及 40K key 跨 4096/16384 边界；Mamba fast path 与保留的
-  full-materialize path 逐项差分；极端 `chunk_size=SIZE_MAX` 也会被安全收敛为单个 suffix range，不发生整数回绕；
+  多 presence word、75 个 required spec 跨 word，以及 40K key 跨 4096/16384 边界；极端
+  `chunk_size=SIZE_MAX` 也会被安全收敛为单个 suffix range，不发生整数回绕；
 - 上述并行/生命周期重点用例 20 轮重复通过；MetaSearcherTest、CacheManagerTest 全量通过；
 - `bazel test --config=release --nocache_test_results //kv_cache_manager/...` 为 106 个测试通过，1 个 GPU-only 测试
   按预期 skip；
@@ -1775,8 +1777,8 @@ notify/join，不改变运行期 heartbeat、ReportEvent 或查询锁范围。
 不再出现长等待。
 
 local compact read 还去掉了一个确定的逐 key 冗余：未配置 revisit histogram 时不再读取
-`last_access_time` atomic；访问时间仍按原语义更新，启用 histogram 时采样逻辑完全不变。另修正混合压测器读取已经
-废弃的 `prefix_match_blocks` 字段所造成的假失败，改为校验当前协议的 `local` 字段。
+`last_access_time` atomic；访问时间仍按原语义更新，启用 histogram 时采样逻辑完全不变。后续兼容性复核确认
+`prefix_match_blocks` 是既有 JSON 字段名，因此保留 tag 2 并恢复该名称，而不是继续暴露临时改名后的 `local`。
 
 最终 Release、pure-local 真实 HTTP 复核：
 
@@ -1817,10 +1819,9 @@ GPU-only 测试按预期跳过。上述 mixed 结果也说明渐进 query 没有
 4. ordinary 与 Mamba visitor 在一个已经读取的 callback range 内，一旦所有候选 host 的单调 prefix stop 都不晚于
    当前 key，就立即停止剩余 location projection。它不取消仍可能决定更短 prefix 的早期 range，也不屏蔽仍在任一
    host 必需区间内的 backend error；
-5. 新增 6000-key、6-host 的确定性随机差分模型，组合 medium、L1.5/L2、全部非 serving 状态、非法 URI、Eagle
-   pop 和 Mamba spec group，把渐进结果逐项与保留的 full-materialize 路径比较。另用 33000 key、两个不同 stop 的
-   host 验证：所有候选 stop 之后的 speculative timeout 可以忽略，但仍处于较长 host prefix 内的同类 timeout 必须
-   原样返回；
+5. 新增 6000-key、6-host 的确定性随机模型，组合 medium、L1.5/L2、全部非 serving 状态、非法 URI、Eagle pop
+   和 Mamba spec group，比较两种查询模式的 local 投影；另用 33000 key、两个不同 stop 的 host 验证：所有候选
+   stop 之后的 speculative timeout 可以忽略，但仍处于较长 host prefix 内的同类 timeout 必须原样返回；
 6. 全量门禁顺便暴露两个测试自身的不确定假设并已固定：MigrationManager 的 capacity partial-failure 用例不再假设
    固定 key 在动态 hash seed 下必然属于不同 shard；两个直接清空 executor queue 的 stale-cache 用例会先停止
    reclaimer supervisor，避免销毁它正在等待的 `packaged_task` 后随机抛出 `broken promise`。两项均只修改测试，
@@ -1842,3 +1843,25 @@ GPU-only 测试按预期跳过。上述 mixed 结果也说明渐进 query 没有
   `bazel test --config=release --nocache_test_results --test_output=errors --jobs=8 //kv_cache_manager/...`：
   106 个可执行测试全部通过，1 个 GPU-only 测试按预期跳过。工作树迁移后复用旧绝对路径 output base 曾导致 7 个
   用例在 0ms 内因 runfiles 缺失失败；独立重建后这些用例全部通过，未把基础设施假失败当作代码缺陷。
+
+### 5.29 2026-08-11 P2P 有界归约与协议兼容性收敛
+
+更新后的基线已经解决 `p2p_host_count=0` 的渐进读取，但 P2P 路径仍为每个 key 构造两份
+`map<host, set<spec>>`。本轮删除这套逐 key 对象图：host 在请求内映射为 dense id，spec group 编码为多 word
+bitmask；普通 prefix 在线维护最终 top-N host 的 peer 交集，Mamba 通过三个有界 ordered pass 保留原来的
+“full group 独立选 prefix peer、state group 统一选 coverage peer”语义。非 local backend 仍只读取一次 compact
+batch，后续 pass 重放该 batch。
+
+同一 Release/O2 二进制、4 query workers、pure-local metadata、1M key、在第 1024 key 制造一个可由 Vineyard
+peer 补齐的 local gap，对更新前后的 `mu-main` 使用完全相同的 benchmark：
+
+| 场景 | 更新前 p50 | 本轮 p50 | 变化 |
+| --- | ---: | ---: | ---: |
+| ordinary P2P | 607.63ms | 155.27ms | -74.4% |
+| Mamba P2P | 627.97ms | 351.25ms | -44.1% |
+| 进程峰值 RSS | 1,342,488KiB | 585,336KiB | -56.4%（约 -739MiB） |
+
+数字来自共享开发机，只用于同机回归，不是线上 SLA。门禁覆盖动态 top-N 淘汰、重复 block key 的 fetched count
+去重、多个 full group 分别选 peer 后合并、非 local backend 单次读取、跨 64 个 spec 的 bitmask、ordered visitor
+窗口/stop，以及 P2P 前缀内硬错误传播。协议侧同时恢复 `HostCacheMatch.prefix_match_blocks` 的既有 JSON 名称；
+`location_spec_names` 保持全局响应投影语义，新 per-key 候选过滤使用 tag 10 的 `per_key_location_spec_names`。

integration_test/meta_service/http_interface_test.py

@@ -243,7 +243,7 @@ def test_event_report_requested_spec_filters_before_peer_selection(self):
                 "query_type": "QT_BATCH_GET",
                 "block_keys": block_keys,
                 "block_mask": {"offset": 0},
-                "location_spec_names": ["linear_1"] * len(block_keys),
+                "per_key_location_spec_names": ["linear_1"] * len(block_keys),
                 "backend_selectors": [{
                     "backend_type": "ST_EVENT_REPORT_L2",
                     "strategy": strategy,
@@ -264,7 +264,7 @@ def test_event_report_requested_spec_filters_before_peer_selection(self):
             "query_type": "QT_BATCH_GET",
             "block_keys": block_keys,
             "block_mask": {"offset": 0},
-            "location_spec_names": ["unknown_spec"] * len(block_keys),
+            "per_key_location_spec_names": ["unknown_spec"] * len(block_keys),
             "backend_selectors": [{
                 "backend_type": "ST_EVENT_REPORT_L2",
                 "strategy": "LSS_V6D_PREFIX",

integration_test/meta_service/test_report_event.py

@@ -1014,7 +1014,7 @@ def test_18_get_host_cache_state_dual_type_prefix_match(self):
             "10.0.0.3:8080": 1,
         }
         actual = {
-            h["host_ip_port"]: int(h["local"])
+            h["host_ip_port"]: int(h["prefix_match_blocks"])
             for h in resp.get("hosts", [])
         }
         self.assertNotIn("p2p_1_hit_count", resp)

integration_test/meta_service/test_report_event_snapshot.py

@@ -1015,7 +1015,7 @@ def test_11_mixed_batch(self):
             "block_cache_keys": [block_key],
         })
         matches = {
-            item["host_ip_port"]: int(item["local"])
+            item["host_ip_port"]: int(item["prefix_match_blocks"])
             for item in host_state.get("hosts", [])
         }
         self.assertEqual(matches.get(host), 1)
@@ -1167,7 +1167,7 @@ def test_16a_heartbeat_timeout_then_recovery(self):
             "block_cache_keys": [block_key],
         })
         target_prefixes = {
-            item["host_ip_port"]: int(item["local"])
+            item["host_ip_port"]: int(item["prefix_match_blocks"])
             for item in target_host_state.get("hosts", [])
         }
         self.assertEqual(target_prefixes.get(host), 1)
@@ -1284,7 +1284,7 @@ def test_16a_heartbeat_timeout_then_recovery(self):
             "block_cache_keys": [block_key],
         })
         hidden_prefixes = {
-            item["host_ip_port"]: int(item["local"])
+            item["host_ip_port"]: int(item["prefix_match_blocks"])
             for item in hidden_host_state.get("hosts", [])
         }
         self.assertEqual(
@@ -1388,7 +1388,7 @@ def test_16a_heartbeat_timeout_then_recovery(self):
             "block_cache_keys": [block_key],
         })
         recovered_prefixes = {
-            item["host_ip_port"]: int(item["local"])
+            item["host_ip_port"]: int(item["prefix_match_blocks"])
             for item in recovered_host_state.get("hosts", [])
         }
         self.assertEqual(recovered_prefixes.get(host), 1)
@@ -1678,7 +1678,7 @@ def test_16_get_host_cache_state(self):
             "10.0.0.3:8080": 1,
         }
         actual = {
-            h["host_ip_port"]: int(h["local"])
+            h["host_ip_port"]: int(h["prefix_match_blocks"])
             for h in resp.get("hosts", [])
         }
         for host, prefix in expected.items():
@@ -1703,7 +1703,7 @@ def test_16_get_host_cache_state(self):
             ["10.0.0.1:8080", "10.0.0.2:8080", "10.0.0.3:8080"],
         )
         large_actual = {
-            item["host_ip_port"]: int(item["local"])
+            item["host_ip_port"]: int(item["prefix_match_blocks"])
             for item in large_hosts
         }
         self.assertEqual(large_actual["10.0.0.1:8080"], 2)
@@ -3231,7 +3231,7 @@ def test_26_unregistered_errors_remain_distinct_from_delta_only_mode(self):
             "block_cache_keys": [25_000_250],
         })
         matches = {
-            item["host_ip_port"]: int(item["local"])
+            item["host_ip_port"]: int(item["prefix_match_blocks"])
             for item in host_state.get("hosts", [])
         }
         self.assertEqual(matches.get(host), 1)
@@ -3422,7 +3422,7 @@ def send_first_delta(writer):
             ],
         })
         matches = {
-            item["host_ip_port"]: int(item["local"])
+            item["host_ip_port"]: int(item["prefix_match_blocks"])
             for item in host_state.get("hosts", [])
         }
         self.assertEqual(matches.get(host), writer_count)
@@ -3536,7 +3536,7 @@ def test_31_first_delete_without_snapshot_creates_reusable_generation(self):
             "block_cache_keys": [block_key],
         })
         empty_matches = {
-            item["host_ip_port"]: int(item["local"])
+            item["host_ip_port"]: int(item["prefix_match_blocks"])
             for item in empty_host_state.get("hosts", [])
         }
         self.assertNotIn(host, empty_matches)
@@ -3582,7 +3582,7 @@ def test_31_first_delete_without_snapshot_creates_reusable_generation(self):
             "block_cache_keys": [block_key],
         })
         visible_matches = {
-            item["host_ip_port"]: int(item["local"])
+            item["host_ip_port"]: int(item["prefix_match_blocks"])
             for item in visible_host_state.get("hosts", [])
         }
         self.assertEqual(visible_matches.get(host), 1)
@@ -4520,7 +4520,7 @@ def test_21_get_host_cache_state_local_scaling(self):
 
             def assert_response(response):
                 prefixes = {
-                    item["host_ip_port"]: int(item["local"])
+                    item["host_ip_port"]: int(item["prefix_match_blocks"])
                     for item in response.get("hosts", [])
                 }
                 self.assertEqual(prefixes.get(host), block_count)
@@ -4557,7 +4557,7 @@ def query_worker(worker_index):
                             f"worker {worker_index}: status={code}, body={body}"
                         )
                     prefixes = {
-                        item["host_ip_port"]: int(item["local"])
+                        item["host_ip_port"]: int(item["prefix_match_blocks"])
                         for item in body.get("hosts", [])
                     }
                     if prefixes.get(host) != block_count:

kv_cache_manager/data_storage/data_storage_backend.h

@@ -25,6 +25,9 @@ class DataStorageBackend {
     virtual double GetStorageUsageRatio(const std::string &trace_id) const = 0;
     inline bool IsOpen() const { return is_open_.load(std::memory_order_relaxed); }
     inline void SetOpen(bool open) { is_open_.store(open, std::memory_order_relaxed); }
+    // Overrides may add notifications, but must delegate to this method. Any
+    // state published by SetAvailable(true) must be initialized before the
+    // release store so IsAvailable()'s acquire load observes it consistently.
     virtual void SetAvailable(bool available) { is_available_.store(available, std::memory_order_release); }
     std::shared_ptr<DataStorageMetricsCollector> GetMetricsCollector() { return metrics_collector_; }
     virtual const StorageConfig &GetStorageConfig() { return config_; }