fix(vllm): recover hybrid KV loads through request-level outcomes

CHANGELOG.md

@@ -2,6 +2,24 @@
 
 ## Unreleased
 
+### v2.27.1 — Request-level hybrid KV load recovery
+
+- Adopt vLLM's native `KVConnectorTransferResults` protocol for hybrid and
+  multi-group loads instead of reporting ambiguous, group-less block IDs.
+- Publish paired receive completion/failure only after native writes and the
+  owning CUDA device are fenced; cover misses, malformed results, exceptions,
+  queue rejection and cancellation without losing or duplicating outcomes.
+- Keep `load_async=false` I/O serialized on the model thread while parking
+  hybrid requests until receive completion. Explicit device synchronization
+  also handles runners that defer the load hook until after a forward pass.
+- Bypass failed external hits for the rest of a request and rebuild retry
+  metadata from fresh allocations, including resumed MRV1 and MRV2 requests.
+- Remove the obsolete hybrid invalid-block scheduler patch. This connector
+  requires native `get_transfer_results` / `failed_recving` support throughout
+  the vLLM runner, executor and scheduler; older runtimes need an engine upgrade.
+- Preserve single full-attention block-level recovery and stored-key layout.
+  No functional C++/server, native ABI or wire-format changes.
+
 ### v2.27.0 — Runtime RDMA providers and native hybrid-state integration
 
 - Include `ibverbs-providers` in the runtime image; exposing RDMA devices does

VERSION

@@ -1 +1 @@
-2.27.0
+2.27.1

docs/CONNECTORS.md

@@ -882,19 +882,23 @@ lsmod | grep nvidia_peermem
 
 #### 混合模型的 KV 加载故障恢复
 
-- vLLM 上游 `_handle_invalid_blocks` 的 `(req_block_ids,) = ...` 解包在多 KV-group
-  模型上必然抛 `ValueError`，导致 EngineCore 死亡。随包提供
-  [混合 invalid-blocks 修复补丁](../integration/vllm/patches/vllm-hybrid-invalid-blocks.patch)：
-  hybrid 请求按 per-group 外层 spec（AttentionSpec 乘 DCP）计算受影响前缀，
-  命中失败时使所有参与 group 的相关 block hash 失效（仅 hash，不释放 DMA 中
-  buffer），`skip_reading_prefix_cache=True` 后复用 `_preempt_request` 回到
-  waiting 队列本地重算；async 请求在 finished_recving 之后由既有
-  `_update_waiting_for_remote_kv` 释放。单 group 请求保留原最长有效前缀与
-  共享 block 优化。fail 策略仅上报受影响请求与 eviction 集合，不做重放。
-  dfkv 调度器同步配合：重放请求查询前丢弃残余 lookup/load_spec 并返回
-  `(0, False)`，不会反复撞同一外部缓存。补丁针对实测镜像
-  `vllm/vllm-openai:glm53-flash` 的 `vllm/v1/core/sched/scheduler.py`，应用前先
-  `git apply --check`；不可盲目覆盖其它引擎版本。
+- 混合/多 group 和单 group 非 full-attention 布局使用 vLLM 原生请求级失败
+  协议，要求引擎提供 `KVConnectorTransferResults.failed_recving`、
+  `KVConnectorOutput.failed_recving` 的完整透传，以及对
+  `WAITING_FOR_REMOTE_KVS` 请求的调度器恢复处理。请升级到具备完整 API 的
+  引擎；不再提供或要求外置 invalid-blocks 调度器补丁，也不按未经验证的
+  版本号推定支持。
+- 这些布局按 request ID 报错，`invalid_block_ids` 留空；只有原生 I/O 和 GPU
+  写入都经过终止 fence 后，才同时报告 `failed_recving` 与 `finished_recving`。
+  `kv_load_failure_policy="recompute"` 时，引擎释放失败分配并重新本地准入；
+  dfkv 丢弃旧 lookup、load spec 和 tracker，按新分配及实际计算前缀重建 SAVE
+  状态。失败请求在本次生命周期内（包括后续抢占）不再查询远端命中，避免同一
+  广告命中不断触发加载—失败循环；最终完成时清理此状态。`"fail"` 策略由引擎
+  终止受影响请求。单 group full-attention 保留既有 block-level 错误恢复。
+- `load_async=False` 只选择串行模型线程 I/O，不取消上述请求的
+  `WAITING_FOR_REMOTE_KVS` 准入。连接器在 GET 前等待先前 GPU 工作完成，
+  GET 后再完成写入 fence；按 runner 调用顺序可能处于前一个 forward 之后，
+  但不与 GPU compute 重叠。失败请求本身在恢复完成前不执行 forward。
 
 ### 3.2 验证
 
@@ -940,7 +944,7 @@ namespace/key 不一致是预期 cold miss。**空环 / MDS 不可达**可直接
 | `batch_concurrency` | `8` | **大池可调高到 ≈ 节点数** | 跨节点 fan-out，**真正的吞吐杠杆**（depth 是平的） |
 | `rail_affinity` | `False` | 多 rank、多 rail 生产设 `true` | 按 vLLM world-group per-host `local_rank` 选择 primary；在 native client 创建前设置每进程独立 rail 环境 |
 | `rail_affinity_fallbacks` | `1` | `1` | 相邻有序 fallback 数；`0`=严格单 rail，超出可用 rail 数时自动收敛 |
-| `load_async` | `True` | 普通 attention 保持 True；hybrid recurrent 模型设 `False` | `False` 在 forward 前同步完成 load，避免 recurrent-state compute 与远端 GPU 写重叠 |
+| `load_async` | `True` | 普通 attention 保持 True；hybrid recurrent 模型设 `False` | `False` 在模型线程串行执行 load，GET 前后均等待 GPU fence，可能位于前一个 forward 之后；hybrid/多 group 和非 full-attention 请求仍进入 `WAITING_FOR_REMOTE_KVS`，不会消费失败 KV。单 group full-attention 保留同步准入。 |
 | `transfer_queue_capacity` | `256` | 保持默认，按压测调 | 每个 worker、每个方向的排队上限（`1..65536`）。满队列时非阻塞拒绝新任务：save 立即释放 finish/free fence，load 标记失败并重算；非法值启动即失败。 |
 | `recv_workers` | `1` | 从 `1` 起压测 | 共享有界 receive queue 的 GET worker 数（`1..32`）；仅在 queue wait 持续升高且后端仍有余量时增加。 |
 | `load_window_keys` | `0`（关闭） | 长上下文 replicated-MLA 按压测设置 | 单次 native GET window 的最大 key 数（`0..65536`）。窗口结果应能放入 `DFKV_NODE_DEDUP_GPU_ARENA_MB`，且在 `DFKV_NODE_DEDUP_WAIT_MS` 内完成，避免 follower rank 超时后重复读取同一批 KV。 |

integration/common/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "dfkv-common"
-version = "2.27.0"
+version = "2.27.1"
 description = "Canonical namespace and pool-key schema shared by dfkv connectors"
 requires-python = ">=3.9"
 

integration/lmcache/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "dfkv-connector"
-version = "2.27.0"
+version = "2.27.1"
 description = "LMCache RemoteConnector for the dfkv KV cache (ctypes over libdfkv.so)"
 readme = "README.md"
 requires-python = ">=3.9"
@@ -14,7 +14,7 @@ authors = [{ name = "Wine93", email = "wine93.info@gmail.com" }]
 # runtime by path (DFKV_LIB / remote_storage_plugin.dfkv.lib). No bundled .so,
 # no CPython extension, so the wheel is platform-independent.
 dependencies = [
-  "dfkv-common==2.27.0",
+  "dfkv-common==2.27.1",
   "lmcache",
   "torch",
 ]

integration/vllm/README.md

@@ -41,6 +41,21 @@ device pointers. Set `DFKV_RDMA=1` in every engine process. Construction rejects
 and closes any native handle whose reported transport is not `rdma`; there is no
 TCP or host-bounce fallback for this connector.
 
+Hybrid/multi-group caches and single-group non-full-attention caches require
+vLLM's native request-level load-failure protocol:
+`KVConnectorTransferResults.failed_recving`, its propagation through
+`KVConnectorOutput.failed_recving`, and scheduler recovery of failed
+`WAITING_FOR_REMOTE_KVS` requests. Upgrade to an engine exposing that complete
+API; no out-of-tree invalid-block scheduler patch is supported or required.
+The connector reports failed request IDs together with receive completion only
+after native I/O and GPU writes are fenced, leaving block-ID errors empty for
+these layouts. With `kv_load_failure_policy="recompute"`, the engine releases the
+failed allocation and retries locally; dfkv bypasses the failed remote source
+for the remainder of that request, including subsequent preemptions. The engine's
+`"fail"` policy terminates the affected request instead.
+The request-level completion fence is mandatory even when the legacy
+`DFKV_GPU_LOAD_FENCE=0` override is present.
+
 ## Environment variables (engine process)
 
 Read by `libdfkv.so` (the C client) and the connector, so set them in **every**
@@ -87,7 +102,7 @@ LMCache connector access logs, so one setting covers every integration. Format:
 | `batch_concurrency` | `0`=auto | client fan-out for batch ops; the real throughput lever (depth is flat). Auto = `min(max(nodes, 8), 32)`: 8-way parallel on single-node, one-per-node on multi-node. Set >0 to pin a fixed value. |
 | `rail_affinity` | `False` | Bind each vLLM worker process to a primary rail selected by world-group local rank; requires an ordered multi-rail `DFKV_RDMA_DEV`. |
 | `rail_affinity_fallbacks` | `1` | Number of ordered neighboring fallback rails when affinity is enabled. `0` keeps strict one-rank/one-rail; values above the available rail count are bounded. |
-| `load_async` | `True` | `True` returns `WAITING_FOR_REMOTE_KVS` and overlaps GPUDirect loads with unrelated model work. `False` performs each requested load synchronously in `start_load_kv`, before the forward pass. Use `False` for hybrid state-cache models when the engine cannot guarantee that remote writes target blocks disjoint from concurrent compute. |
+| `load_async` | `True` | Controls I/O execution, not hybrid admission. `True` overlaps GPUDirect loads with unrelated model work. `False` executes serialized loads on the model thread, fences preceding GPU work before GET, and fences completion before returning; depending on the runner, this may occur after the preceding forward. Hybrid/multi-group and non-full-attention requests still enter `WAITING_FOR_REMOTE_KVS` in either mode, so failed KV is never consumed by their forward pass. Single-group full-attention `False` loads retain synchronous admission. Use `False` for hybrid state-cache models when remote writes cannot safely overlap compute. |
 | `transfer_queue_capacity` | `256` | Maximum queued requests in each direction (`1..65536`). All receive workers consume one shared receive queue of this capacity; capacity is not multiplied by `recv_workers`. Submission is non-blocking: a full queue rejects new saves as completed (releasing finish/free fences) and rejects new loads as load errors (forcing recompute), so overload cannot grow memory or pin blocks indefinitely. Invalid or out-of-range values abort connector construction. |
 | `recv_workers` | `1` | Receive/load worker count (`1..32`). Workers consume the shared bounded receive queue and may execute independent native GETs concurrently. Invalid, boolean, or out-of-range values abort connector construction. |
 | `load_window_keys` | `0` (disabled) | Maximum keys per native GET window (`0..65536`). Use a value whose worst-case result bytes fit inside the node-dedup GPU arena. Windowing lets follower ranks consume published results before the dedup wait deadline instead of re-fetching a large replicated-MLA batch. |

integration/vllm/pyproject.toml

@@ -4,12 +4,12 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "dfkv-vllm"
-version = "2.27.0"
+version = "2.27.1"
 description = "Direct vLLM KVConnectorBase_V1 connector for dfkv (GPUDirect RDMA, no LMCache)"
 requires-python = ">=3.12"
 # vllm + torch are provided by the runtime image; not pinned here. The connector
 # itself is pure Python (ctypes over libdfkv.so), so there is no native build.
-dependencies = ["dfkv-common==2.27.0"]
+dependencies = ["dfkv-common==2.27.1"]
 
 # Telemetry is opt-in: the OTel SDK is only needed when DFKV_METRICS_ENABLED /
 # DFKV_TRACE_ENABLED is set. Without this extra the connector stays dependency-

integration/vllm/src/dfkv_vllm/connector.py

@@ -27,6 +27,7 @@
     KVConnectorBase_V1,
     KVConnectorMetadata,
     KVConnectorRole,
+    KVConnectorTransferResults,
     SupportsHMA,
 )
 from vllm.distributed.kv_transfer.kv_connector.v1.metrics import (
@@ -244,6 +245,8 @@ def reset_cache(self) -> bool | None:
             self._finish_call()
 
     def update_connector_output(self, connector_output: KVConnectorOutput):
+        assert self.connector_scheduler is not None
+        self.connector_scheduler.update_connector_output(connector_output)
         kv_cache_events = connector_output.kv_cache_events
         if not kv_cache_events or not isinstance(
             kv_cache_events, DfkvStoreKVEvents
@@ -319,8 +322,8 @@ def handle_preemptions(
     def start_load_kv(self, forward_context: ForwardContext, **kwargs: Any) -> None:
         self._begin_call()
         try:
-            # Loads are issued in get_finished() for compute overlap. Synchronous
-            # loads required by this step are submitted here before forward.
+            # Async loads are issued during result collection. Inline loads
+            # fence prior kernels before touching destination blocks.
             assert self.connector_worker is not None
             metadata = self._get_connector_metadata()
             assert isinstance(metadata, DfkvStoreConnectorMetadata)
@@ -353,20 +356,20 @@ def save_kv_layer(
     def wait_for_save(self):
         self._begin_call()
         try:
-            # get_finished submits stores and fences mutable/windowed sources.
+            # Result collection submits stores and fences mutable/windowed sources.
             return
         finally:
             self._finish_call()
 
-    def get_finished(
+    def get_transfer_results(
         self, finished_req_ids: set[str]
-    ) -> tuple[set[str] | None, set[str] | None]:
+    ) -> KVConnectorTransferResults:
         self._begin_call()
         try:
             assert self.connector_worker is not None
             metadata = self._get_connector_metadata()
             assert isinstance(metadata, DfkvStoreConnectorMetadata)
-            return self.connector_worker.get_finished(finished_req_ids, metadata)
+            return self.connector_worker.get_transfer_results(finished_req_ids, metadata)
         finally:
             self._finish_call()
 

integration/vllm/src/dfkv_vllm/data.py

@@ -31,6 +31,18 @@
 # they must cold-miss rather than enter the current restore path.
 VLLM_RAW_LAYOUT = b"vllm-multiwr-v4"
 
+
+def requires_request_level_loads(kv_cache_config) -> bool:
+    """Use request identities when physical groups cannot share a block grid."""
+    from vllm.v1.kv_cache_interface import FullAttentionSpec
+
+    groups = getattr(
+        kv_cache_config, "transfer_groups", kv_cache_config.kv_cache_groups
+    )
+    return len(groups) > 1 or any(
+        not isinstance(group.kv_cache_spec, FullAttentionSpec) for group in groups
+    )
+
 def key_diagnostic_label(key: bytes) -> str:
     """Return the standard non-reversible diagnostic label for a store key."""
     try: