Add ReportEvent snapshot URI versioning

docs/README.md

@@ -3,6 +3,7 @@
 ### 设计文档
 - [模块架构与关联关系](design/module_architecture.md) - 各模块职责、依赖方向、控制流与数据流，附 Mermaid 图
 - [基本概念](design/basic_concepts.md) - Storage、Instance Group、Instance、Block、CacheLocation 等核心概念
+- [ReportEvent Snapshot URI 版本方案](design/report_event_snapshot_uri_version.md) - Snapshot 全量覆盖、URI 版本、MightExist 过滤与后台清理
 - [高可用与选主机制](design/ha_leader_elector.md) - HA 架构、LeaderElector 状态机、CoordinationBackend、Leader 发现
 
 ### 开发文档

docs/api/meta_service.md

@@ -101,6 +101,58 @@ curl -g -vvv -X POST http://localhost:6382/api/removeCache \
 }'
 ```
 
+## Report Event
+
+`reportEvent` is the cache-subscriber ingestion API. A subscriber normalizes
+engine-specific cache signals into block-level events:
+
+- RTP-LLM subscriber: report full host/location snapshots with `EVENT_BLOCK_SNAPSHOT`.
+- vLLM subscriber: map KV events to ordered `EVENT_BLOCK_ADD`,
+  `EVENT_BLOCK_DELETE`, and full-clear snapshots.
+
+`EVENT_BLOCK_SNAPSHOT` is authoritative for one `host_ip_port + medium`.
+KVCM writes an internal snapshot version into each reported URI, filters old
+versions through `MightExist`, and deletes stale versions asynchronously in a
+background scan. Callers should not set the internal `kvcm_*` URI params
+themselves.
+
+```bash
+curl -g -vvv -X POST http://localhost:6382/api/reportEvent \
+  -H "Content-Type: application/json" \
+  -H "Accept: application/json" \
+  -d '{
+    "trace_id": "trace_id_131",
+    "instance_id": "test_instance",
+    "host_ip_port": "192.168.2.1:8080",
+    "storage_type": "ST_VINEYARD",
+    "events": [
+      {
+        "event_type": "EVENT_NODE_REGISTER",
+        "node_register": {
+          "mediums": ["gpu"]
+        }
+      },
+      {
+        "event_type": "EVENT_BLOCK_SNAPSHOT",
+        "block_snapshot": {
+          "medium": "gpu",
+          "blocks": [
+            {
+              "block_key": "123",
+              "specs": [
+                {
+                  "name": "full_attention:group=0:tp=0",
+                  "uri": "vineyard://192.168.2.1:8080/gpu/123?size=4096"
+                }
+              ]
+            }
+          ]
+        }
+      }
+    ]
+}'
+```
+
 ## Trim Cache
 ```bash
 curl -g -vvv -X POST http://localhost:6382/api/trimCache \
@@ -129,4 +181,4 @@ curl -g -vvv -X POST http://localhost:6382/api/getCacheMeta \
     },
     "detail_level": 1
 }'
-```
+```

docs/design/module_architecture.md

@@ -4,7 +4,7 @@
 
 > **维护提示**：当模块的职责、依赖方向或调用关系发生变化，或新增/删除模块时，请同步更新本文档与文末的 Mermaid 图，并同步更新 [AGENTS.md](../../AGENTS.md) 中的缩略图。
 
-相关文档：[基本概念](basic_concepts.md)、[高可用与选主机制](ha_leader_elector.md)、[配置指南](../configuration.md)、[优化器文档](../optimizer.md)。
+相关文档：[基本概念](basic_concepts.md)、[ReportEvent Snapshot URI 版本方案](report_event_snapshot_uri_version.md)、[高可用与选主机制](ha_leader_elector.md)、[配置指南](../configuration.md)、[优化器文档](../optimizer.md)。
 
 ---
 

docs/design/report_event_snapshot_uri_version.md

@@ -0,0 +1,105 @@
+# ReportEvent Snapshot URI 版本方案
+
+## 1. 背景
+
+`EVENT_BLOCK_SNAPSHOT` 表示 subscriber 上报某个 `instance_id + host_ip_port + medium` 当前完整拥有的 block 集合。PR230 的方案是在 meta 层维护持久化的 `location_id -> block_key` 反向索引，请求路径基于该索引做精确 diff：新增缺失 block，删除未上报 block。
+
+本方案保留 PR230 的 proto 形状，但不引入 location 反向索引。snapshot 全量覆盖的判断通过 URI 版本完成：每次 full snapshot 分配一个递增版本，把版本写入本次上报 block 的 `LocationSpec.uri`，再用 `MightExist` 和后台扫描过滤旧版本 location。
+
+## 2. 目标与非目标
+
+目标：
+
+- `ReportEvent` proto 与 PR230 保持一致，包括 `EVENT_BLOCK_SNAPSHOT` 和 `BlockSnapshotEventParams`。
+- 避免在请求路径扫描全量 CacheMeta，也避免新增 Redis set 作为 location 反向索引。
+- snapshot 写入成功后，旧版本 location 立即不再被查询命中。
+- 后台扫描异步删除旧版本 metadata，避免旧 location 长期残留。
+- KVCM 重启恢复 CacheMeta 后，能从已有 URI 回填 event backend 内存中的最大版本号。
+
+非目标：
+
+- 不在本次实现 mamba-state 的精细匹配。
+- 不保证 snapshot request 内同步删除所有旧 metadata；删除由后台扫描完成。
+- 不把 URI 版本暴露成用户查询接口。
+
+## 3. 版本维度
+
+版本按如下维度维护：
+
+```text
+instance_id + host_ip_port + medium -> snapshot_version
+```
+
+`medium` 仍然来自 proto payload，用来区分同一 host 下不同可独立对账的 cache namespace。版本不是只按 host 维护，否则同一个 host 上不同 instance 或 medium 会互相误伤。
+
+写入 CacheMeta 的 URI 会附加三个参数：
+
+```text
+kvcm_instance_id=<instance_id>
+kvcm_medium=<medium>
+kvcm_snapshot_version=<version>
+```
+
+示例：
+
+```text
+vineyard://192.168.1.1:8080/mem/100?kvcm_instance_id=test_instance&kvcm_medium=mem&kvcm_snapshot_version=2&size=1
+```
+
+## 4. 写入流程
+
+`EVENT_BLOCK_SNAPSHOT` 的处理流程：
+
+1. 校验整个 snapshot payload，包括 `medium`、`block_key` 和每个 `LocationSpec.uri`。
+2. 从 event backend 分配下一个 snapshot version，但先不发布为最新版本。
+3. 给本次 snapshot 中所有 URI 写入版本参数。
+4. 复用现有 `MetaSearcher::BatchUpsertLocations` 写入 reported blocks。
+5. 只有当该 snapshot event 对应的写入全部成功时，才 commit 版本为最新版本。
+6. commit 后提交后台任务扫描 CacheMeta，删除旧版本 location。
+
+这样可以避免 snapshot 写入部分失败时立刻把上一版全量结果判为过期。已写入但未 commit 的更高版本不会被判为 stale；下一次成功 snapshot 会自然覆盖它。
+
+普通 `EVENT_BLOCK_ADD` 不分配新版本。如果同一 `instance_id + host + medium` 已经有 committed snapshot version，block add 会沿用当前版本写入 URI，表示它属于当前 snapshot generation。
+
+空 snapshot 是合法的。它会分配并 commit 一个新版本，然后通过查询过滤和后台扫描让该 host/medium 的旧版本全部失效。
+
+## 5. 查询与后台清理
+
+查询路径仍然从 CacheMeta 读取 `CacheLocationMap`。在 `CheckLocDataExistFunc` 中增加 snapshot 版本判断：
+
+- 如果 location id 可以解析出 `host_ip_port + medium`，并且 event backend 中存在该 scope 的最新版本。
+- 如果 location URI 中的 `kvcm_snapshot_version` 小于最新版本，则该 location 视为不存在。
+- 对同一 scope 下没有版本参数的 legacy ReportEvent location，在已经存在最新 snapshot version 后，也视为旧版本。
+- 版本判断通过后，再调用原有 `DataStorageManager::Exist(..., fastpath=true)`，即 backend 的 `MightExist`。
+
+后台清理由 `CleanupStaleSnapshotLocations` 完成：
+
+1. 扫描当前 instance 的 CacheMeta。
+2. 只检查目标 `storage_type + host_ip_port + medium` 的 location。
+3. 用同一套版本规则判断 stale location。
+4. 调用 `BatchDeleteLocations` 删除旧 metadata。
+
+因此读路径负责快速过滤，后台扫描负责最终清理；两者不依赖新的 location 反向索引。
+
+## 6. 重启恢复
+
+event backend 的版本表是内存状态。KVCM 恢复 instance 的 meta indexer 后，会执行一次版本恢复扫描：
+
+1. 遍历已有 CacheMeta location。
+2. 解析 URI 中的 `kvcm_instance_id`、`kvcm_medium`、`kvcm_snapshot_version`。
+3. 和 location id 解析出的 `host_ip_port + medium` 做一致性校验。
+4. 用最大观察版本回填 event backend。
+
+恢复完成后，下一次 full snapshot 会继续从最大版本 `+1` 分配，而不是从 1 重新开始。
+
+## 7. 与 PR230 方案的差异
+
+本方案不维护 `location_id -> block_key` 反向索引，也不在 snapshot 请求路径计算 `reported_keys - existing_keys` / `existing_keys - reported_keys`。
+
+取而代之：
+
+- 新 snapshot 写入最新版本的 reported blocks。
+- 未上报的旧 block 保留在 CacheMeta 中，但其 URI 版本小于最新版本，查询时立即过滤。
+- 后台扫描随后删除这些旧版本 location。
+
+优点是改动面小，不要求 meta backend 增加持久 set 接口；代价是删除旧 metadata 不是请求同步完成，短时间内 CacheMeta 会保留被版本规则屏蔽的旧 location。

kv_cache_manager/data_storage/event_reporting_backend.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <cstdint>
 #include <functional>
 #include <map>
 #include <string>
@@ -10,6 +11,10 @@
 
 namespace kv_cache_manager {
 
+inline constexpr const char *KVCM_SNAPSHOT_INSTANCE_PARAM = "kvcm_instance_id";
+inline constexpr const char *KVCM_SNAPSHOT_MEDIUM_PARAM = "kvcm_medium";
+inline constexpr const char *KVCM_SNAPSHOT_VERSION_PARAM = "kvcm_snapshot_version";
+
 class EventReportingBackend {
 public:
     using CleanupCallback =
@@ -31,8 +36,26 @@ class EventReportingBackend {
     virtual bool IsCleanupCallbackSet() const = 0;
 
     virtual std::string BuildLocationId(const std::string &medium, const std::string &host_ip_port) const = 0;
+    virtual bool ParseLocationId(const std::string &location_id,
+                                 std::string &out_medium,
+                                 std::string &out_host_ip_port) const = 0;
     virtual std::string HostSuffix(const std::string &host_ip_port) const = 0;
 
+    virtual uint64_t AllocateSnapshotVersion(const std::string &instance_id,
+                                             const std::string &host_ip_port,
+                                             const std::string &medium) = 0;
+    virtual void CommitSnapshotVersion(const std::string &instance_id,
+                                       const std::string &host_ip_port,
+                                       const std::string &medium,
+                                       uint64_t version) = 0;
+    virtual void ObserveSnapshotVersion(const std::string &instance_id,
+                                        const std::string &host_ip_port,
+                                        const std::string &medium,
+                                        uint64_t version) = 0;
+    virtual uint64_t GetSnapshotVersion(const std::string &instance_id,
+                                        const std::string &host_ip_port,
+                                        const std::string &medium) const = 0;
+
     virtual DataStorageType GetStorageType() const = 0;
 
     virtual std::string GetProtocol() const = 0;