[manager] move EventReport cleanup into background GC

Author: shaohuaxi

docs/README.md

@@ -8,6 +8,7 @@
 - [高可用与选主机制](design/ha_leader_elector.md) - HA 架构、LeaderElector 状态机、CoordinationBackend、Leader 发现
 - [CacheReclaimer 异步删除与过度逐出优化](design/cache_reclaimer_async_delete.md) - 异步删除生命周期、in-flight credit、反压与无进展退避
 - [后台扫描 GC](design/cache_garbage_collector.md) - 基于 authoritative cursor 的后台全量巡检；V1 清理长期 orphan WRITING 和普通 SERVING storage-missing，并提供无副作用读取、精确值条件 CAS 与 HA 生命周期
+- [EventReport 主动回收纳入后台扫描 GC](design/event_report_background_gc.md) - 以可合并 cleanup intent 取代 per-event 全 Instance 扫描，统一 Snapshot generation 与 down host metadata 回收
 
 ### 开发文档
 - [开发指南](develop/README.md) - 开发者入门指南和开发环境配置

docs/configuration.md

@@ -134,6 +134,9 @@ kvcm.cache_gc.scan_batch_size=256
 kvcm.cache_gc.orphan_writing_grace_period_ms=86400000
 # GC 在途删除请求硬上限；默认 2。一个慢请求只占一个槽位，全部槽位占满后暂停扫描。
 kvcm.cache_gc.max_inflight_delete_requests=2
+# 迁移期开关：将 EventReport Snapshot/HOST_DOWN 回收路由到 GC cleanup intent lane；
+# 默认关闭，开启时要求 kvcm.cache_gc.enabled=true，配置变更需重启生效。
+kvcm.cache_gc.event_report_cleanup_enabled=false
 
 # 可选值有dummy，local，logging，kmonitor；若不配置或配置为空，默认使用local
 kvcm.metrics.reporter_type=local

docs/design/cache_garbage_collector.md

@@ -3,7 +3,7 @@
 | 项目 | 内容 |
 |---|---|
 | 状态 | V1 已实现并完成相关单测、E2E 与性能冒烟；基线已包含异步删除（#234）、分层存储迁移（#209）和精确值条件删除（#233） |
-| 更新时间 | 2026-07-30 |
+| 更新时间 | 2026-08-14 |
 | 涉及模块 | `manager`、`meta`、`data_storage`、`config`、`metrics`、`service` |
 | 历史参考 | [PR #184](https://github.com/alibaba/tair-kvcache/pull/184) |
 
@@ -101,7 +101,7 @@ V1 实现以下六项能力：
 4. 动态或大规模并发窗口、多维 bytes/Group 配额、Future deadline 和持久化任务。
 5. `WriteLocationManager` 三元组索引、settling guard，以及 Reclaimer/Finish 竞态修复。
 6. `CLS_DELETING` 自动恢复和 storage version/epoch。
-7. EventReport reconciliation、Block TTL 和闲置 Instance 清理。
+7. EventReport reconciliation、Block TTL 和闲置 Instance 清理不属于本文定义的 GC V1；其中 EventReport 已作为独立扩展接入同一后台 GC，详见 [EventReport 主动回收纳入后台扫描 GC](event_report_background_gc.md)。
 8. 查询链路 fast-submit 重构。
 9. GC lease/generation、跨进程 cursor 或 pending 恢复。
 10. data storage I/O timeout/cancel 和 CAS 后补偿状态机。
@@ -605,11 +605,11 @@ Block TTL 需要先定义 TTL 写入、刷新、authoritative `expire_at`、并
 4. GC task 持久化、lease/generation 和跨进程恢复；
 5. data storage I/O timeout/cancel；
 6. authoritative time source；
-7. EventReport snapshot/event 的后台 reconciliation；其 metadata-only、版本和所有权语义不能复用普通 storage 的物理删除规则。
+7. EventReport snapshot/event 的后台 reconciliation 已作为独立扩展设计和交付；其 metadata-only、版本和所有权语义不复用普通 storage 的物理删除规则，详见 [EventReport 主动回收纳入后台扫描 GC](event_report_background_gc.md)。
 
 ### 11.6 建议交付顺序
 
 1. **GC V1 PR**：本文定义的 full scan、固定 WRITING 与普通 SERVING storage-missing 判定、精确值条件删除、有界 Future/pending、leader 生命周期、指标和测试。
 2. **公共基础设施跟进（按需）**：若现网验证确认 Redis 命令缺少有限返回上界，再独立补齐 connect/auth/command/reconnect timeout 和坏连接淘汰。
 3. **独立正确性 PR**：WriteLocationManager Instance 隔离和 Reclaimer/Finish settling race。
-4. **后续 PR**：根据性能和业务优先级分别接入 pacing/index、显式三态 probe、TTL、EventReport 和 DELETING reconciliation。
+4. **后续 PR**：根据性能和业务优先级分别接入 pacing/index、显式三态 probe、TTL 和 DELETING reconciliation；EventReport reconciliation 按独立扩展文档交付。

docs/design/event_report_background_gc.md

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

docs/design/module_architecture.md

@@ -395,6 +395,8 @@ snapshot replace + commit/abort
 
 ## 11. 限流与清理
 
+本节保留 legacy fallback 的行为说明；启用 `kvcm.cache_gc.event_report_cleanup_enabled` 后，系统按 [EventReport 主动回收纳入后台扫描 GC](event_report_background_gc.md) 把 per-event 全 Instance 扫描替换为可合并 cleanup intent。两条路径的 Snapshot、generation、lifecycle fencing 和 metadata-only 契约保持一致。
+
 `EventReportStorageSpec.snapshot_min_interval_ms` 提供 per-reporter 最小 snapshot 间隔，默认
 30 秒。完整维度是：
 

docs/design/report_event_snapshot_uri_version.md

@@ -214,8 +214,117 @@ def test_missing_serving_data_is_collected_without_query(self):
             "missing SERVING metadata should be removed and become writable",
         )
 
+    def test_event_report_host_down_is_collected_by_intent_lane(self):
+        """HOST_DOWN is reconciled by the metadata-only intent lane."""
+        self.worker_manager.stop_worker(0)
+        self.assertTrue(
+            self.worker_manager.start_worker(
+                0,
+                **{
+                    "kvcm.cache_gc.enabled": "true",
+                    "kvcm.cache_gc.event_report_cleanup_enabled": "true",
+                    "kvcm.cache_gc.scan_interval_ms": 25,
+                    "kvcm.cache_gc.round_pause_ms": 100,
+                    "kvcm.cache_gc.scan_batch_size": 8,
+                    "kvcm.cache_gc.orphan_writing_grace_period_ms": self._GRACE_MS,
+                    "kvcm.cache_gc.max_inflight_delete_requests": 2,
+                },
+            )
+        )
+        self._admin_client.close()
+        self._client.close()
+        self._admin_client, self._client = self._get_manager_clients()
+
+        event_storage_name = "cache_gc_event_report"
+        self._admin_client.add_storage(
+            {
+                "trace_id": f"{self._trace_id}_event_storage",
+                "storage": {
+                    "global_unique_name": event_storage_name,
+                    "storage_type": "ST_EVENT_REPORT_L2",
+                    "event_report": {
+                        "heartbeat_timeout_ms": 60_000,
+                        "cleanup_grace_ms": 60_000,
+                        "liveness_check_interval_ms": 1_000,
+                    },
+                    "check_storage_available_when_open": False,
+                },
+            }
+        )
+        self._create_topology(
+            event_report_storage_candidates=[event_storage_name]
+        )
+        instance_id = "cache_gc_event_report_instance"
+        self._register_instance(instance_id)
+        host = "192.168.80.1:8080"
+        location_uri = f"vineyard://{host}/mem?source=cache_gc_e2e"
+        self._client.report_event(
+            {
+                "trace_id": f"{self._trace_id}_event_add",
+                "instance_id": instance_id,
+                "host_ip_port": host,
+                "storage_type": "ST_EVENT_REPORT_L2",
+                "events": [
+                    {
+                        "event_type": "EVENT_NODE_REGISTER",
+                        "node_register": {"mediums": ["mem"]},
+                    },
+                    {
+                        "event_type": "EVENT_BLOCK_ADD",
+                        "block_add": {
+                            "block_key": str(self._block_key),
+                            "medium": "mem",
+                            "specs": [{"name": "tp0", "uri": location_uri}],
+                        },
+                    },
+                ],
+            }
+        )
+        visible = self._client.get_cache_location(
+            {
+                "trace_id": f"{self._trace_id}_event_visible",
+                "instance_id": instance_id,
+                "query_type": "QT_BATCH_GET",
+                "block_keys": [self._block_key],
+                "block_mask": {"offset": 0},
+            }
+        )
+        self.assertTrue(visible.get("locations"))
+
+        self._client.report_event(
+            {
+                "trace_id": f"{self._trace_id}_event_down",
+                "instance_id": instance_id,
+                "host_ip_port": host,
+                "storage_type": "ST_EVENT_REPORT_L2",
+                "events": [
+                    {"event_type": "EVENT_HOST_DOWN", "host_down": {}}
+                ],
+            }
+        )
+        self._wait_metric_at_least(
+            "cache_gc.event_report_delete_location_count",
+            1,
+            tags={"reason": "down_host", "status": "deleted"},
+            timeout_s=10,
+        )
+        self.assertGreaterEqual(
+            self._metric_value("cache_gc.event_report_scan_batch_count"), 1
+        )
+        self.assertEqual(
+            0,
+            self._metric_value(
+                "cache_gc.event_report_intent_count",
+                tags={"reason": "down_host"},
+            ),
+        )
+        self.assertEqual(0, self._metric_value("cache_gc.delete_target_count"))
+
     def _create_topology(
-        self, meta_storage_type="dummy", data_storage_type="nfs"
+        self,
+        meta_storage_type="dummy",
+        data_storage_type="nfs",
+        event_report_storage_candidates=None,
     ):
         metadata_uri = (
             f"file://{self.get_workdir()}/cache_gc_metadata"
@@ -247,6 +356,9 @@ def _create_topology(
                 "instance_group": {
                     "name": self._group_name,
                     "storage_candidates": [self._storage_name],
+                    "event_report_storage_candidates": (
+                        event_report_storage_candidates or []
+                    ),
                     "global_quota_group_name": "cache_gc_quota",
                     "max_instance_count": 8,
                     "quota": {

integration_test/cache_garbage_collector/cache_garbage_collector_test.py

@@ -61,6 +61,10 @@ def finish_write_cache(self, data, check_response=True):
         """Finish writing cache data"""
         return self._make_api_request('/api/finishWriteCache', data, check_response)
 
+    def report_event(self, data, check_response=True):
+        """Report external cache lifecycle events"""
+        return self._make_api_request('/api/reportEvent', data, check_response)
+
     def remove_cache(self, data, check_response=True):
         """Remove cache data for specified block keys"""
         return self._make_api_request('/api/removeCache', data, check_response)

integration_test/meta_service/http_interface_test.py

@@ -500,6 +500,17 @@ class Cache {
             const std::string_view &key, ObjectPtr obj, size_t charge, const CacheItemHelper *helper)> & /*callback*/) {
     }
 
+    // Applies a callback to one existing entry without marking it as a hit or
+    // changing its LRU position. The callback runs while the cache shard is
+    // locked and returns the entry charge delta caused by an in-place update.
+    // Callers must keep the callback bounded and must not call back into this
+    // Cache. Returns false when the key is absent or unsupported.
+    virtual bool ApplyToEntryNoTouch(
+        const std::string_view & /*key*/,
+        const std::function<ssize_t(ObjectPtr obj, size_t charge, const CacheItemHelper *helper)> & /*callback*/) {
+        return false;
+    }
+
     // Insert a mapping from key->object only if the key does not already exist.
     // Returns EC_OK on successful insertion, EC_EXIST if the key is already
     // present, or other error codes on failure (e.g. EC_NOSPC).
@@ -777,6 +788,12 @@ class CacheWrapper : public Cache {
         target_->ApplyToSingleShard(shard_id, callback);
     }
 
+    bool ApplyToEntryNoTouch(
+        const std::string_view &key,
+        const std::function<ssize_t(ObjectPtr obj, size_t charge, const CacheItemHelper *helper)> &callback) override {
+        return target_->ApplyToEntryNoTouch(key, callback);
+    }
+
     void StartAsyncLookup(AsyncLookupHandle &async_handle) override { target_->StartAsyncLookup(async_handle); }
 
     void WaitAll(AsyncLookupHandle *async_handles, size_t count) override { target_->WaitAll(async_handles, count); }

kv_cache_manager/common/cache/advanced_cache.h

@@ -497,6 +497,52 @@ void LRUCacheShard::LookupBatch(const std::string_view *keys,
     }
 }
 
+bool LRUCacheShard::ApplyToEntryNoTouch(
+    const std::string_view &key,
+    uint32_t hash,
+    const std::function<ssize_t(Cache::ObjectPtr obj, size_t charge, const Cache::CacheItemHelper *helper)> &callback) {
+    std::lock_guard<std::mutex> l(mutex_);
+    LRUHandle *e = table_.Lookup(key, hash);
+    if (e == nullptr) {
+        return false;
+    }
+    assert(e->InCache());
+    const bool in_lru = !e->HasRefs();
+    const ssize_t delta = callback(e->value, e->total_charge, e->helper);
+    if (delta > 0) {
+        const size_t increase = static_cast<size_t>(delta);
+        e->total_charge += increase;
+        usage_ += increase;
+        if (in_lru) {
+            lru_usage_ += increase;
+            if (e->InHighPriPool()) {
+                high_pri_pool_usage_ += increase;
+            } else if (e->InLowPriPool()) {
+                low_pri_pool_usage_ += increase;
+            }
+            MaintainPoolSize();
+        }
+    } else if (delta < 0) {
+        size_t decrease = static_cast<size_t>(-delta);
+        decrease = std::min(decrease, e->total_charge);
+        decrease = std::min(decrease, usage_);
+        e->total_charge -= decrease;
+        usage_ -= decrease;
+        if (in_lru) {
+            assert(lru_usage_ >= decrease);
+            lru_usage_ -= decrease;
+            if (e->InHighPriPool()) {
+                assert(high_pri_pool_usage_ >= decrease);
+                high_pri_pool_usage_ -= decrease;
+            } else if (e->InLowPriPool()) {
+                assert(low_pri_pool_usage_ >= decrease);
+                low_pri_pool_usage_ -= decrease;
+            }
+        }
+    }
+    return true;
+}
+
 bool LRUCacheShard::Ref(LRUHandle *e) {
     std::lock_guard<std::mutex> l(mutex_);
     // To create another reference - entry must be already externally referenced.