执行摘要

• 一句话：PushConnector 记录远端活跃时间以清理过期引擎
• 推荐动作：值得精读。该 PR 虽小，但体现了分布式连接管理中一个典型陷阱：继承的清理逻辑需要正确接入活跃信号源才能生效。阅读重点：_do_start_push_kv 中刷新位置的选择及其注释，以及两个针对性测试如何用桩构造方式隔离真实 NIXL。对于想理解 vLLM KV connector 可靠性和 NIXL push 模式的工程师，这是很好的入门案例。

功能与动机

PR body 明确说明：PushConnector 继承了 base_worker 中清理 stale remote 数据结构、防止 CPU 泄漏的逻辑（源自 PR #44424），但自己没有记录每个远端引擎的最后活动时间，因此无法触发该清理流程。issue #48633 的 S1 条目指出 D 引擎在 scale up/down 后，P 侧 _remote_agents 和 NIXL agent 可能残留，导致连接泄漏。本 PR 通过记录 last activity 让 TTL eviction 真正生效。

实现拆解

实现分三步：

1. 在 push 热路径刷新活跃时间（vllm/distributed/kv_transfer/kv_connector/v1/nixl/push_worker.py）：在 _do_start_push_kv 中，当 _ensure_handshake 返回 None（即引擎已连接、WRITE 直接执行）时，新增 self._engine_last_active[decode_engine_id] = time.perf_counter()。关键点是：_ensure_handshake 只在握手建立时刷新活跃时间，对已连接引擎的持续 push 不会刷新，若不在此处记录，正在活跃传输的引擎也会被 TTL 误判为过期而回收。该行放在 WRITE 提交之前，与 pull 模式 start_load_kv 的语义对齐。
2. 修正 base_worker 注释（vllm/distributed/kv_transfer/kv_connector/v1/nixl/base_worker.py）：_cleanup_remote_engine 中原注释声称 push P 侧引擎不参与 eviction，现改为「刚完成的握手可能尚未记录活动时间，因此容忍缺失条目」，仅注释变更，逻辑不变，因为 pop(engine_id, None) 本就容忍缺失。
3. 补充单元测试（tests/v1/kv_connector/unit/test_nixl_push_connector.py）：新增 _eviction_worker 辅助构造器，创建具备 _ensure_handshake/_cleanup_remote_engine 所需全部字段的 worker 桩；新增 test_active_push_refreshes_engine_last_active 验证活跃 push 会刷新 last_active，避免被误回收；新增 test_stale_engine_evicted_on_push 验证超过 TTL 的 D 引擎在 push 触发握手时被删除，并调用 remove_remote_agent。测试均通过 stub _xfer_blocks_for_req 记录 WRITE 调用，不依赖真实 NIXL。

关键文件：

• vllm/distributed/kv_transfer/kv_connector/v1/nixl/push_worker.py（模块 KV连接器；类别 source；类型 core-logic；符号 _do_start_push_kv）: 核心逻辑变更：在 push 直写路径刷新远端引擎最后活跃时间，使 TTL eviction 对 push 模式生效。
• vllm/distributed/kv_transfer/kv_connector/v1/nixl/base_worker.py（模块 KV连接器；类别 source；类型 core-logic；符号 _cleanup_remote_engine）: 注释修正：说明 push 模式下 last_active 现在也会被记录，pop 容忍缺失的语义不变。
• tests/v1/kv_connector/unit/test_nixl_push_connector.py（模块 测试；类别 test；类型 test-coverage；符号 _eviction_worker, test_active_push_refreshes_engine_last_active, test_stale_engine_evicted_on_push）: 新增 eviction 场景的测试桩与两个关键测试，验证活跃刷新与过期回收。

关键符号：_do_start_push_kv, _cleanup_remote_engine, _eviction_worker, test_active_push_refreshes_engine_last_active, test_stale_engine_evicted_on_push

关键源码片段

vllm/distributed/kv_transfer/kv_connector/v1/nixl/push_worker.py

核心逻辑变更：在 push 直写路径刷新远端引擎最后活跃时间，使 TTL eviction 对 push 模式生效。

# vllm/distributed/kv_transfer/kv_connector/v1/nixl/push_worker.py
# 摘取自 _do_start_push_kv 的关键路径：处理已连接引擎的直写分支
​
# 尝试建立 / 复用与 decode 引擎的握手；返回 Future 表明握手尚未完成
fut = self._ensure_handshake(
    decode_engine_id,
    registration_data["decode_host"],
    registration_data["decode_port"],
    registration_data["decode_tp_size"],
)
if fut is not None:
    # 后台 executor 正在握手，完成后的 WRITE 会通过 _deferred_push_inbox 重放；
    # 这里不能立即刷新 last_active，因为引擎此时尚未确认可用
    fut.add_done_callback(_on_handshake)
    return
​
# 引擎已连接：WRITE 直接执行。
# 关键点：_ensure_handshake 仅在首次握手时刷新 last_active，
# 持续 push 不会刷新；若不在此处更新时间戳，活跃引擎也会被
# _evict_stale_engines 按 TTL 误回收（pull 模式在 start_load_kv 中同样处理）
self._engine_last_active[decode_engine_id] = time.perf_counter()
​
# 以下为原有逻辑：构造 ReqMeta 并提交 NIXL WRITE
logical_local = self._as_grouped_block_ids(local_block_ids)
logical_remote = self._as_grouped_block_ids(remote_block_ids)
physical_local = self._logical_to_kernel_block_ids(
    logical_local, self._physical_blocks_per_logical_kv_block
)
push_meta = ReqMeta(
    local_block_ids=logical_local,
    local_physical_block_ids=physical_local,
    tp_size=self.world_size,
    remote=RemoteMeta(
        block_ids=logical_remote,
        host="",
        port=0,
        engine_id=decode_engine_id,
        request_id=decode_request_id,
    ),
)
self._xfer_blocks_for_req(req_id=request_id, meta=push_meta)

tests/v1/kv_connector/unit/test_nixl_push_connector.py

新增 eviction 场景的测试桩与两个关键测试，验证活跃刷新与过期回收。

# tests/v1/kv_connector/unit/test_nixl_push_connector.py
# 新增的 eviction 场景测试：验证 push 路径下的 TTL 行为
​
​
def _eviction_worker(engine_ttl: float) -> NixlPushConnectorWorker:
    """构造一个具备 base _ensure_handshake 清理路径所需字段的 push worker。"""
    w = _StubWriterWorker.fresh()
    w._engine_ttl = engine_ttl
    w._engine_last_active = {}
    w._engine_clock_offset = {}
    w._handshake_lock = threading.RLock()
    # _cleanup_remote_engine 回收引擎时会触碰这些结构
    w.nixl_wrapper = MagicMock()
    w.dst_xfer_side_handles = {}
    w.kv_caches_base_addr = {}
    w.dst_num_blocks = {}
    w.tp_mappings = {}
    w.transfer_topo = None
    w._logical_to_kernel_block_ids = lambda blocks, ratio: blocks
    w.writes = []
    w._xfer_blocks_for_req = lambda req_id, meta: w.writes.append(req_id)
    return w
​
​
def test_active_push_refreshes_engine_last_active():
    """活跃 push 必须刷新 last_active，否则正在传输的引擎会被误回收。"""
    w = _eviction_worker(engine_ttl=3600.0)
    # 已连接 D 引擎 -> _ensure_handshake 返回 None，WRITE 内联执行
    w._remote_agents["decode-engine"] = {(0, 0): "agent-decode"}
    stale = time.perf_counter() - 5.0
    w._engine_last_active["decode-engine"] = stale
​
    _real_do_start_push_kv(w, "req", ([1, 2, 3],), _registration_data("req"))
​
    assert w._engine_last_active["decode-engine"] > stale
    assert w.writes == ["req"]
​
​
def test_stale_engine_evicted_on_push():
    """push 触发握手清理：静默超过 TTL 的 D 引擎被回收。"""
    w = _eviction_worker(engine_ttl=30.0)
    w._remote_agents["D-old"] = {(0, 0): "agent-D-old"}
    w._engine_last_active["D-old"] = time.perf_counter() - 10_000.0
    # 被 push 的目标引擎本身已连接
    w._remote_agents["decode-engine"] = {(0, 0): "agent-decode"}
​
    _real_do_start_push_kv(w, "req", ([1, 2, 3],), _registration_data("req"))
​
    assert "D-old" not in w._remote_agents
    assert "D-old" not in w._engine_last_active
    w.nixl_wrapper.remove_remote_agent.assert_called_once_with("agent-D-old")

评论区精华

该 PR 的 review 过程非常简单：维护者 chaunceyjiang 直接批准（LGTM），没有 review comments。claude[bot] 自动评论说明 fork PR 默认关闭自动 review，可手动触发。因此没有实质性的技术争论或设计权衡讨论；核心设计决策（在 _do_start_push_kv 而非 _ensure_handshake 中刷新）由 PR 作者在代码注释和测试描述中自行说明。

• 暂无高价值评论线程

风险与影响

• 风险：风险点包括：
  1. 活跃时间刷新位置依赖：刷新逻辑仅放在 _ensure_handshake 返回 None 的分支（已连接直写路径）。若未来新增其它 push 入口（如 deferred push 重放、不同调用方），可能遗漏刷新导致误回收。当前测试已覆盖直写路径，但未覆盖握手回调 _on_handshake 后的 deferred push 路径——该路径通过 _deferred_push_inbox 最终也会进入 _do_start_push_kv，因此间接安全。
  2. 并发安全：_engine_last_active 由主线程（_do_start_push_kv 常在主线程执行）写入，而 _evict_stale_engines 在握手线程或主线程执行时遍历，均为 dict 单操作，CPython 下基本安全，但缺少显式锁，极端情况下多线程同时 push 到同一引擎可能产生非原子读改写。
  3. TTL 误判：刷新时间点是 WRITE 提交前而非完成后，若 WRITE 实际耗时超过 TTL 阈值（默认可能很大），理论上仍可能被回收；但 NIXL WRITE 是异步提交，风险很低。
  4. 行为回归：base_worker 注释变更不影响逻辑；测试为新增，无既有测试改动。整体风险可控。
• 影响：影响范围集中在 NixlPushMode（PD 分离 push 场景）的 P 侧 worker：
  • 对用户：修复长时间运行下 D 引擎 scale down 后 P 侧 agent 泄漏问题，提升大规模部署稳定性，减少 CPU 内存占用。
  • 对系统：启用 base_worker 已有的 TTL eviction，使 _evict_stale_engines 在 push 模式下真正生效，与 pull 模式的行为对齐。
  • 对团队：为 issue #48633 的可靠性路线图补齐 S1 关键项，后续可在同一框架上扩展更细粒度的 liveness 控制。改动小（源码 +3 行），风险低。
  • 风险标记：TTL 误判风险, push 热路径变更, 并发写无显式锁

关联脉络

• PR #51100 [Bugfix] Fix Mamba all-mode CPU offload boundary alignment: 同属 kv-connector v1 可靠性修复，与本次 push 模式 TTL 清理共同提升 KV 传输稳定性。
• PR #49644 [Feat][Core] Add disk offloading support to SimpleCPUOffloadConnector: 同为 KV connector 方向的演进，扩展连接器生命周期管理能力。