执行摘要

• 一句话：阻止双向模式下 P 读取 D 端过期 KV blocks
• 推荐动作：值得精读，尤其是时钟同步策略、安全裕度设计以及如何在不引入过多配置项的情况下保证正确性。对于分布式 KV 传输贡献者，本 PR 展示了以最小侵入性解决数据一致性问题的方法。

功能与动机

来自 PR 描述：当前多轮双向模式下，D 为 KV blocks 设置固定 TTL，但 heartbeat 不会延长该 TTL，因此 blocks 过期后可被其他请求复用。后续 turn 无法识别该 blocks 是否有效，导致读取到乱码。本 PR 通过让 D 提交 blocks 过期时间，P 在读取前校验，解决该问题。

实现拆解

变更分五个核心步骤：

1. 版本 Bump 与元数据扩展：在 metadata.py 中将 NIXL_CONNECTOR_VERSION 从 4 提升到 5，并在 RemoteMeta 中添加 blocks_expiry_time: float | None 字段，在 add_new_req_to_recv 中从 kv_transfer_params 读取 remote_blocks_expiry_time。
2. Handshake 时钟同步：修改 base_worker.py 中的 _nixl_handshake，在发送握手消息前记录 time.perf_counter()，收到回复后解析第二个帧（D 端的 perf_counter），计算 RTT 并取最小 RTT 样本对应的时钟偏移 best_offset。同时，pull_scheduler.py 中的 request_finished 在 D 节点上导出本地 blocks_expiry_time 为 remote_blocks_expiry_time。
3. 时钟偏移写入：将计算出的偏移存入 base_worker.py 新增的 _engine_clock_offset 字典，该字典在 done_callback 中（以线程安全方式）写入，而非直接在 handshake 后台线程中写入。
4. 过期校验逻辑：在 pull_worker.py 中新增 _is_turn2_read_expired 方法，将 D 端 blocks_expiry_time 减去本地估计的时钟偏移，并与当前 perf_counter 加上安全裕度（5 秒）比较；如果过期或即将过期则返回 True。在 _read_blocks_for_req 开头检查该条件，若为真则记录 xfer_stats 并调用 _handle_failed_transfer 拒绝本次读取。
5. 双向模式门控与测试：整个过期检查逻辑仅在 _bidirectional_kv_xfer_enabled 为 True（通过配置 bidirectional_kv_xfer 启用）时执行。同时，test_bidirectional_kv_transfer.py 新增 4 个测试用例覆盖偏移、过期、安全裕度、缺失 deadline 等场景，并调整已有辅助函数以适配新版本 handshake 签名（返回双值）。
   配套变更包括 base_scheduler.py 中 handshake listener 回显 perf_counter，push_worker.py 适配签名，test_nixl_connector.py 和 test_nixl_connector_hma.py 适配参数变化，以及 docs/design/nixl_kv_cache_lease.md 更新文档说明。

关键文件：

• tests/v1/kv_connector/unit/test_bidirectional_kv_transfer.py（模块 双向传输；类别 test；类型 test-coverage；符号 test_turn2_deadline_gate, test_turn2_full_prefix_hit_with_expired_deadline_skips_gate, test_d_node_request_finished_exports_blocks_expiry_time, test_handshake_listener_appends_perf_counter_frame）: 新增4个测试用例覆盖 turn-2 deadline 校验的各类场景（过期、偏移导致过期、安全裕度内、远未过期、缺失 deadline），确保 P 端正确拒绝或放行读取。同时调整辅助函数以适配 handshake 返回 signature 变化。
• vllm/distributed/kv_transfer/kv_connector/v1/nixl/base_worker.py（模块 分布式；类别 source；类型 core-logic；符号 done_callback）: 核心改动：启用双向模式配置标志；新增 _engine_clock_offset 字典；修改 _nixl_handshake 以测量 RTT 和时钟偏移；在 done_callback 中安全写入偏移。
• vllm/distributed/kv_transfer/kv_connector/v1/nixl/pull_worker.py（模块 分布式；类别 source；类型 core-logic；符号 _is_turn2_read_expired）: 新增过期检查核心函数 _is_turn2_read_expired，并在 _read_blocks_for_req 开头调用；只有双向模式启用时才执行。
• vllm/distributed/kv_transfer/kv_connector/v1/nixl/pull_scheduler.py（模块 调度器；类别 source；类型 core-logic）: 在 request_finished 中，当 is_d_node 为 True 时，导出 blocks_expiry_time 作为 remote_blocks_expiry_time 传递给 P。
• vllm/distributed/kv_transfer/kv_connector/v1/nixl/metadata.py（模块 元数据；类别 source；类型 data-contract）: 版本 bump 4->5，添加 RemoteMeta.blocks_expiry_time 字段，并在 add_new_req_to_recv 中解析。

关键符号：_is_turn2_read_expired, _nixl_handshake, _nixl_handshake_listener, request_finished, add_new_req_to_recv, done_callback

关键源码片段

tests/v1/kv_connector/unit/test_bidirectional_kv_transfer.py

新增4个测试用例覆盖 turn-2 deadline 校验的各类场景（过期、偏移导致过期、安全裕度内、远未过期、缺失 deadline），确保 P 端正确拒绝或放行读取。同时调整辅助函数以适配 handshake 返回 signature 变化。

@pytest.mark.parametrize(
    ("offset", "expiry_delta", "expect_declined"),
    [
        # 已过期：D 端 deadline 比当前时间早 10s → 应拒绝
        pytest.param(0.0, -10.0, True, id="expired"),
        # 时钟偏移致过期：P 认为 D 快 100s，D 的 deadline 还有 50s → 相对 P 已过期
        pytest.param(100.0, 50.0, True, id="offset_makes_expired"),
        # 在 5s 安全裕度内 → 视为可能过期，拒绝
        pytest.param(0.0, 2.0, True, id="near_expiry_within_margin"),
        # 未来很远 → 有效
        pytest.param(0.0, 1000.0, False, id="valid_far_deadline"),
        # 无 deadline（旧版本节点） → 跳过，正常读取
        pytest.param(0.0, None, False, id="missing_expiry_field"),
    ],
)
@patch("vllm.distributed.kv_transfer.kv_connector.v1.nixl.base_worker.NixlWrapper",
       FakeNixlWrapper)
def test_turn2_deadline_gate(dist_init, offset, expiry_delta, expect_declined):
    """校验 P 在 turn-2 读取时根据 deadline 拒绝或放行的行为。"""
    connector, worker = _make_connector_with_fake_worker()
    worker._engine_clock_offset[_REMOTE] = offset # 注入 P 对 D 的时钟偏差
    # 构造 deadline：如果 expiry_delta 为 None 则缺失 deadline
    expiry_time = None if expiry_delta is None else time.perf_counter() + expiry_delta
    meta = NixlConnectorMetadata()
    params = {
        "do_remote_prefill": False,
        "do_remote_decode": True,
        "remote_block_ids": ([20, 21],),
        "remote_engine_id": FakeNixlConnectorWorker.REMOTE_ENGINE_ID,
        "remote_request_id": "decode-req",
        "remote_host": "localhost",
        "remote_port": 1234,
        "remote_tp_size": 1,
    }
    if expiry_time is not None:
        params["remote_blocks_expiry_time"] = expiry_time
    meta.add_new_req_to_recv(
        request_id="req",
        local_block_ids=([10, 11],),
        kv_transfer_params=params,
    )
    _do_load_kv(connector, meta)
    # 检查统计数据：应该拒绝（增加 num_kv_expired_reqs）或正常
    expected_expired_count = [1] if expect_declined else [0]
    assert worker.xfer_stats.data["num_kv_expired_reqs"] == expected_expired_count

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

新增过期检查核心函数 _is_turn2_read_expired，并在 _read_blocks_for_req 开头调用；只有双向模式启用时才执行。

# 安全裕度（秒），用于吸收时钟偏移误差和读取延迟
_KV_BLOCKS_EXPIRY_SAFETY_MARGIN = 5.0
​
class NixlPullConnectorWorker(NixlBaseConnectorWorker):
    # ... 其他代码
​
    def _is_turn2_read_expired(self, meta: ReqMeta) -> bool:
        """判断 D 端缓存的 blocks 对本次 turn-2 读取是否已（接近）过期。"""
        assert meta.remote is not None
        blocks_expiry_time = meta.remote.blocks_expiry_time
        # 若 deadline 不存在（如旧版本或 router 未转发），则正常读取
        if blocks_expiry_time is None or not meta.local_physical_block_ids:
            return False
        # 从当前记录的偏移中查找对应 D 引擎的时钟偏移
        clock_offset = self._engine_clock_offset[meta.remote.engine_id]
        # 将 D 域的 deadline 转换为本地域时间
        deadline = blocks_expiry_time - clock_offset
        # 当前时间加上裕度 >= deadline 视为过期
        return time.perf_counter() + _KV_BLOCKS_EXPIRY_SAFETY_MARGIN >= deadline
​
    def _read_blocks_for_req(self, req_id: str, meta: ReqMeta):
        assert meta.remote is not None and self.transfer_topo is not None
        engine_id = meta.remote.engine_id
        self._engine_last_active[engine_id] = time.perf_counter()
        # 双向模式延迟退出：若 blocks 已过期则拒绝读取
        if self._bidirectional_kv_xfer_enabled and self._is_turn2_read_expired(meta):
            logger.warning(
                "Declining expired remote read for %s from engine %s.",
                req_id,
                engine_id,
            )
            self.xfer_stats.record_kv_expired_req()
            self._handle_failed_transfer(req_id, None)
            return
        # ... 后续正常读取流程

评论区精华

• 安全裕度是否可配置：NickLucche 认为应减少配置项，使用硬编码 5 秒即可。最终移除该配置项，改为模块级常量 _KV_BLOCKS_EXPIRY_SAFETY_MARGIN。
• RTT 最小化策略：NickLucche 要求明确为什么取最小 RTT 样本，解释为同一集群内 rank 间传输延迟近似，高 RTT 为噪声，使用最小值可使时钟偏移估计更准确。最终添加了注释说明。
• 时钟偏移写入位置：NickLucche 指出不应在 handshake 线程中直接写入 _engine_clock_offset，应返回偏移值并在 done_callback 中写入，因为 _engine_clock_offset 在主线程读取，需避免竞争。最终调整。
• 多部分消息断言：NickLucche 建议 assert len(reply_parts) == 2，因为版本 bump 已破坏向后兼容，可以直接要求两个帧。最终添加了断言。

• 逻辑提取与门控：NickLucche 建议将过期检查提取为独立方法 _is_turn2_read_expired，并仅在 _bidirectional_kv_xfer_enabled 启用时调用，以减少对常规流程的认知负担。作者最终提取了该方法。

  • 安全裕度是否应暴露为配置 (design): 不暴露，使用模块级常量 _KV_BLOCKS_EXPIRY_SAFETY_MARGIN = 5.0。
  • RTT 最小化策略的设计理由 (design): 添加注释说明选择最小 RTT 的原因。
  • 时钟偏移写入位置应放在 done_callback 中 (correctness): 修改为在 done_callback 中写入偏移，确保主线程读取时一致。
  • 过期检查应提取为独立方法并双边门控 (testing): 提取为独立方法，门控在双向模式配置下。

风险与影响

• 风险：
  • 时钟同步误差：RTT 最小化策略仍可能因网络波动导致偏移估计偏差，但 5 秒安全裕度提供了缓冲，且测试覆盖了多种偏移场景，风险较低。
  • 性能影响：handshake 增加一次 perf_counter 回复，但仅在连接时发生，对运行时无影响；过期检查在每次 read 调用时增加一次浮点比较，可忽略。
  • 不兼容性：版本号 bump 4→5 意味着旧版 NIXL 节点无法通信，但由于组件通常一起升级，影响可控。
  • 潜在竞态：done_callback 与主线程可能并发访问 _engine_clock_offset，但主线程读取时假设 handshake 已完成，当前使用 RLock 保护 handshake futures，读取时未加显式锁，需要确认无竞态。
• 影响：
  • 用户：多轮对话质量显著提升，不再因读取过期 block 出现乱码；需启用 bidirectional_kv_xfer 新配置。
  • 系统：handshake 阶段增加一次额外消息，但频率极低；验证逻辑新增在 _read_blocks_for_req 中，对吞吐影响可忽略。
  • 团队：简化了配置接口（未暴露 safety margin），使代码更易维护；单元测试覆盖了关键路径，降低后续回归风险。
  • 风险标记：时钟同步误差, 安全裕度硬编码, 版本兼容性破坏, 潜在竞态

关联脉络

• 暂无明显关联 PR