fix(vllm): support legacy and native request failure protocols

CHANGELOG.md

@@ -2,6 +2,29 @@
 
 ## Unreleased
 
+### v2.27.2 — Legacy and native vLLM compatibility
+
+- Restore one connector package for engines with the legacy `get_finished`
+  lifecycle and engines with native `get_transfer_results` support.
+- On legacy V1 engines, carry failed request identities through supported
+  worker metadata and enter the existing scheduler recovery/error path only
+  after all participating workers finish. No fabricated block IDs or installed
+  engine source-file patches are used.
+- Preserve originating-child routing through nested `MultiConnector` metadata;
+  do not retain or replay other connectors' historical metadata.
+- Register the external connector's canonical name so MultiConnector can
+  reconstruct cross-process statistics without terminating API output handling.
+  Preserve an operator's existing registration of the same class.
+- Keep serialized hybrid loads parked and fenced, quarantine failed external
+  hits, and preserve cancellation/deferred-free ownership. Native engines
+  continue to use their native request-failure protocol.
+- Register CPU connector and worker lifecycle regressions in CTest, and cover
+  old/native scheduler and executor behavior in their actual engine runtimes.
+- Correct v2.27.1's native-only connector compatibility restriction. Legacy
+  support requires the V1 worker-metadata, completion aggregation and scheduler
+  recovery hooks; arbitrary V0 engines or partial protocol backports are not
+  implied. No functional C++ changes, ABI/wire changes or stored-key migration.
+
 ### v2.27.1 — Request-level hybrid KV load recovery
 
 - Adopt vLLM's native `KVConnectorTransferResults` protocol for hybrid and

CMakeLists.txt

@@ -264,6 +264,14 @@ if(DFKV_BUILD_TESTS)
               -p test_client_ranks.py)
     set_tests_properties(python_vllm_client_ranks PROPERTIES
       ENVIRONMENT "PYTHONPATH=${CMAKE_CURRENT_SOURCE_DIR}/integration/common/src:${CMAKE_CURRENT_SOURCE_DIR}/integration/vllm/src")
+    add_test(NAME python_vllm_connector_lifecycle
+      COMMAND ${PYTHON3}
+              ${CMAKE_CURRENT_SOURCE_DIR}/test/python/test_dfkv_vllm_connector.py
+      WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR})
+    add_test(NAME python_vllm_worker_lifecycle
+      COMMAND ${PYTHON3}
+              ${CMAKE_CURRENT_SOURCE_DIR}/test/python/test_dfkv_vllm_worker.py
+      WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR})
     # Shared telemetry layer: pure-python, no build/lib needed (OTel-push test
     # self-skips when the SDK is absent). Includes the vendored-copy drift guard.
     add_test(NAME python_telemetry

VERSION

@@ -1 +1 @@
-2.27.1
+2.27.2

docs/CONNECTORS.md

@@ -882,14 +882,24 @@ lsmod | grep nvidia_peermem
 
 #### 混合模型的 KV 加载故障恢复
 
-- 混合/多 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`。
+- 一个 connector wheel 自动适配两类引擎，不按未经验证的版本号推定支持。
+  混合/多 group 和单 group 非 full-attention 布局始终按 request ID 报错：
+  **原生路径**通过 `get_transfer_results()` 返回
+  `KVConnectorTransferResults.failed_recving`，由引擎透传
+  `KVConnectorOutput.failed_recving` 并恢复等待请求，不安装调度器桥接。
+  **旧引擎路径**通过 `get_finished()` 收集同一份 worker 结果，将失败 ID
+  保存在 `build_connector_worker_meta()` 返回的 worker metadata 中；多次
+  poll 不会丢失尚未上报的失败，原生路径不会重复发送这份 metadata。
+- 旧引擎必须具备 `KVConnectorWorkerMetadata.aggregate()` 及其传输、executor
+  跨 rank receive-completion 聚合，以及调度器 `update_from_output`、
+  `_handle_invalid_blocks`、等待请求恢复和 load-failure policy 接口。
+  仅在旧引擎的 scheduler-role connector 使用请求级布局时，自动安装经过
+  能力检查的进程内桥接；缺失所需能力时启动明确报错。无需修改已安装的
+  vLLM 源文件，也无需手动打引擎补丁，不影响普通 block error 或其他连接器。
+  桥接跨 step 保留失败，直到聚合后的 `finished_recving` 确认所有 rank 完成，
+  才按请求身份交给既有恢复/终止路径，绝不伪造或扫描混合布局的 block ID。
+- 这些布局的 worker `invalid_block_ids` 留空；只有原生 I/O 和 GPU
+  写入都经过终止 fence 后，才发布失败与 receive completion。
   `kv_load_failure_policy="recompute"` 时，引擎释放失败分配并重新本地准入；
   dfkv 丢弃旧 lookup、load spec 和 tracker，按新分配及实际计算前缀重建 SAVE
   状态。失败请求在本次生命周期内（包括后续抢占）不再查询远端命中，避免同一

integration/common/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "dfkv-common"
-version = "2.27.1"
+version = "2.27.2"
 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.1"
+version = "2.27.2"
 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.1",
+  "dfkv-common==2.27.2",
   "lmcache",
   "torch",
 ]

integration/vllm/README.md

@@ -41,18 +41,43 @@ 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.
+Hybrid/multi-group caches and single-group non-full-attention caches report
+request-level load failures. One connector wheel automatically selects the
+engine's protocol:
+
+- **Native engines:** `get_transfer_results()` returns
+  `KVConnectorTransferResults.failed_recving` alongside receive completion.
+  The engine propagates `KVConnectorOutput.failed_recving` and performs its
+  native scheduler recovery. No scheduler bridge is installed.
+- **Legacy engines:** `get_finished()` uses the same worker outcome collection;
+  `build_connector_worker_meta()` carries failed request IDs through
+  `KVConnectorWorkerMetadata.aggregate()`. For request-level layouts, the
+  scheduler-role connector installs a capability-checked, in-process bridge.
+  It retains failures across steps until executor-aggregated `finished_recving`
+  confirms all ranks are done, then routes request identities through the
+  engine's existing recovery/error path without inventing block IDs.
+
+The legacy bridge requires worker-metadata aggregation/transport, all-rank
+receive-completion aggregation, and the scheduler's `update_from_output`,
+`_handle_invalid_blocks`, waiting-request recovery and load-failure policy
+hooks. Missing bridge capabilities fail explicitly at startup. It does not
+modify installed vLLM source files or require a manually applied engine patch;
+compatibility is based on these capabilities, not a blanket version cutoff.
+Single-group full-attention retains its existing block-level recovery.
+
+Direct and nested `MultiConnector` compositions preserve failure routing to
+the originating dfkv child without replaying other children's metadata.
+Importing the external module also registers `DfkvStoreConnector` by name so
+MultiConnector can reconstruct serialized statistics in the API process.
+An existing registration of the same class is accepted; a conflicting class
+binding is not silently replaced.
+
+Failed request IDs and receive completion are published only after native I/O
+and GPU writes are fenced. These layouts leave block-ID errors empty. 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.
 

integration/vllm/pyproject.toml

@@ -4,12 +4,12 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "dfkv-vllm"
-version = "2.27.1"
+version = "2.27.2"
 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.1"]
+dependencies = ["dfkv-common==2.27.2"]
 
 # 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

@@ -23,11 +23,11 @@
     KVConnectorKVEvents,
     KVEventAggregator,
 )
+from vllm.distributed.kv_transfer.kv_connector.factory import KVConnectorFactory
 from vllm.distributed.kv_transfer.kv_connector.v1.base import (
     KVConnectorBase_V1,
     KVConnectorMetadata,
     KVConnectorRole,
-    KVConnectorTransferResults,
     SupportsHMA,
 )
 from vllm.distributed.kv_transfer.kv_connector.v1.metrics import (
@@ -45,9 +45,19 @@
 from vllm.v1.outputs import KVConnectorOutput
 from vllm.v1.request import Request
 
-from .data import DfkvStoreConnectorMetadata, VLLM_RAW_LAYOUT
+from .data import (
+    DfkvStoreConnectorMetadata,
+    VLLM_RAW_LAYOUT,
+    requires_request_level_loads,
+)
 from .metrics import DfkvStoreConnectorStats, DfkvStorePromMetrics
 from .scheduler import DfkvStoreScheduler
+from .transfer_protocol import (
+    HAS_NATIVE_TRANSFER_RESULTS,
+    LegacyReceiveFailures,
+    TransferResults,
+    install_legacy_failure_bridge,
+)
 from .worker import DfkvStoreWorker
 
 logger = init_logger(__name__)
@@ -144,11 +154,17 @@ def __init__(
         self._inflight_calls = 0
         self._shutdown = False
         self._shutdown_complete = False
+        self._legacy_failed_recving: set[str] = set()
 
         self.connector_scheduler: DfkvStoreScheduler | None = None
         self.connector_worker: DfkvStoreWorker | None = None
 
         if role == KVConnectorRole.SCHEDULER:
+            if (
+                not HAS_NATIVE_TRANSFER_RESULTS
+                and requires_request_level_loads(kv_cache_config)
+            ):
+                install_legacy_failure_bridge()
             self.connector_scheduler = DfkvStoreScheduler(
                 vllm_config, kv_cache_config
             )
@@ -361,15 +377,42 @@ def wait_for_save(self):
         finally:
             self._finish_call()
 
-    def get_transfer_results(
+    def _collect_transfer_results(self, finished_req_ids: set[str]) -> TransferResults:
+        assert self.connector_worker is not None
+        metadata = self._get_connector_metadata()
+        assert isinstance(metadata, DfkvStoreConnectorMetadata)
+        return self.connector_worker.get_transfer_results(finished_req_ids, metadata)
+
+    def get_transfer_results(self, finished_req_ids: set[str]) -> TransferResults:
+        """Return native outcomes without also publishing legacy metadata."""
+        self._begin_call()
+        try:
+            return self._collect_transfer_results(finished_req_ids)
+        finally:
+            self._finish_call()
+
+    def get_finished(
         self, finished_req_ids: set[str]
-    ) -> KVConnectorTransferResults:
+    ) -> tuple[set[str], set[str]]:
+        """Adapt one outcome snapshot to the legacy completion/metadata hooks."""
         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_transfer_results(finished_req_ids, metadata)
+            results = self._collect_transfer_results(finished_req_ids)
+            with self._shutdown_condition:
+                self._legacy_failed_recving.update(results.failed_recving)
+            return results.finished_sending, results.finished_recving
+        finally:
+            self._finish_call()
+
+    def build_connector_worker_meta(self) -> LegacyReceiveFailures | None:
+        self._begin_call()
+        try:
+            with self._shutdown_condition:
+                failed_recving = self._legacy_failed_recving
+                if not failed_recving:
+                    return None
+                self._legacy_failed_recving = set()
+            return LegacyReceiveFailures(failed_recving=failed_recving)
         finally:
             self._finish_call()
 
@@ -432,6 +475,7 @@ def shutdown(self) -> None:
                     errors.append(error)
         finally:
             with self._shutdown_condition:
+                self._legacy_failed_recving.clear()
                 self._shutdown_complete = True
                 self._shutdown_condition.notify_all()
 
@@ -461,3 +505,16 @@ def build_prom_metrics(
         return DfkvStorePromMetrics(
             vllm_config, metric_types, labelnames, per_engine_labelvalues
         )
+
+
+# MultiConnector reconstructs cross-process statistics by class name rather
+# than module path. Register the same external class for that supported route.
+try:
+    KVConnectorFactory.register_connector(
+        "DfkvStoreConnector", __name__, "DfkvStoreConnector"
+    )
+except ValueError:
+    if KVConnectorFactory.get_connector_class_by_name(
+        "DfkvStoreConnector"
+    ) is not DfkvStoreConnector:
+        raise

integration/vllm/src/dfkv_vllm/scheduler.py

@@ -28,6 +28,7 @@
     RequestTracker,
     requires_request_level_loads,
 )
+from .transfer_protocol import failed_requests_from_output
 from .worker import (
     LookupKeyClient,
 )
@@ -190,7 +191,7 @@ def update_connector_output(self, output: KVConnectorOutput) -> None:
         """Prevent a failed remote source from being re-admitted on retry."""
         if not self.request_level_loads:
             return
-        for req_id in output.failed_recving or ():
+        for req_id in failed_requests_from_output(output):
             # A receive can finish after an aborted request's final metadata.
             # Such an output must not recreate scheduler-side request state.
             if req_id not in self._unfinished_request_ids: