[Bugfix][EC Connector] Fail the request, not the engine, when a remote encoding cannot arrive

tests/v1/ec_connector/unit/test_scheduler_nixl_consumer.py

@@ -1,7 +1,10 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+import time
 import uuid
 
+import pytest
+
 import vllm.distributed.ec_transfer.ec_connector.cpu.scheduler as sched_mod
 from tests.v1.ec_connector.unit.utils import create_ec_vllm_config
 from vllm.distributed.ec_transfer.ec_connector.cpu.ec_shared_region import (
@@ -156,7 +159,13 @@ def test_no_params_is_ready(monkeypatch):
     s.shutdown()
 
 
-def test_tombstoned_read_discards_and_blocks_retry(monkeypatch):
+def test_tombstoned_read_fails_the_request(monkeypatch):
+    """A failed remote read must fail the request, not admit it.
+
+    The media reached the producer only, so an admitted request has no
+    embedding and no way to compute one: the model would receive an empty
+    multimodal batch and assert.
+    """
     s = _consumer_sched(monkeypatch)
     fake = _FakeSession()
 
@@ -177,16 +186,13 @@ def _fake_start(mm_hash, info, size):
     fake._results.tombstoned.add("h1")
     s._sessions[("h", 1)] = fake
 
-    # Step 2: poll discards the entry and records a tombstone; the tombstone
-    # is consumed by admit so the request proceeds to local compute.
-    assert s.ensure_cache_available(req, 0) is True
+    # Step 2: the entry is discarded and the request is reported unschedulable.
+    assert s.ensure_cache_available(req, 0) is False
     assert s._cache.get("h1") is None
     assert "h1" not in s._in_flight
-    s.build_connector_meta(scheduler_output=None)
-
-    # Step 3: tombstone was consumed, so a fresh transfer starts again.
-    assert s.ensure_cache_available(req, 0) is False
-    assert "h1" in s._in_flight
+    assert s.get_unrecoverable_requests() == {"r1"}
+    # Drained by the read, so the scheduler cannot abort it twice.
+    assert s.get_unrecoverable_requests() == set()
     s.shutdown()
 
 
@@ -225,14 +231,77 @@ def _fake_start(mm_hash, info, size):
     s.shutdown()
 
 
-def test_size_mismatch_skips_transfer(monkeypatch):
+def test_size_mismatch_fails_the_request(monkeypatch):
+    """A bad announcement is terminal: there is no second source to try."""
     s = _consumer_sched(monkeypatch)
     # Advertised size disagrees with pos.length * hidden_dim * element_size.
     bad = {"h1": {"peer_host": "h", "peer_port": 1, "size_bytes": 999}}
     req = _Request([_Feature("h1", 1)], params=bad)
-    assert s.ensure_cache_available(req, 0) is True
+    assert s.ensure_cache_available(req, 0) is False
     assert "h1" not in s._in_flight
     assert s._cache.get("h1") is None
+    assert s.get_unrecoverable_requests() == {"r1"}
+    s.shutdown()
+
+
+@pytest.mark.parametrize(
+    "announced",
+    [
+        {"peer_host": "h", "peer_port": 1, "size_bytes": None},
+        {"peer_host": "h", "peer_port": 1, "size_bytes": "big"},
+        {"peer_host": "h", "peer_port": 1},
+        123,
+    ],
+    ids=["null-size", "text-size", "no-size", "not-a-mapping"],
+)
+def test_unusable_announcement_fails_the_request(monkeypatch, announced):
+    """`ec_transfer_params` arrives on the request, so it may be anything.
+
+    Raising here would propagate out of Scheduler.schedule() and take the
+    engine down, which is the failure this whole path exists to prevent.
+    """
+    s = _consumer_sched(monkeypatch)
+    req = _Request([_Feature("h1", 1)], params={"h1": announced})
+    admitted = s.ensure_cache_available(req, 0)
+    if announced == 123:
+        # Not a mapping at all: indistinguishable from "never remote", so the
+        # encoder runs locally rather than the request failing.
+        assert admitted is True
+        assert s.get_unrecoverable_requests() == set()
+    else:
+        assert admitted is False
+        assert s.get_unrecoverable_requests() == {"r1"}
+    assert "h1" not in s._in_flight
+    s.shutdown()
+
+
+def test_deferral_budget_is_per_request(monkeypatch):
+    """A request must not inherit an earlier one's clock for a shared item.
+
+    Two requests needing the same encoding hit the same transient condition;
+    the newcomer gets its own budget, so the one that has actually waited too
+    long is the one that fails.
+    """
+    s = _consumer_sched(monkeypatch)
+    # A not-ready orphan entry defers every request that needs it.
+    assert s._cache.alloc("h1", 1) is not None
+    params = _params("h1", 1)
+    old = _Request([_Feature("h1", 1)], params=params, req_id="old")
+    new = _Request([_Feature("h1", 1)], params=params, req_id="new")
+
+    now = [1000.0]
+    monkeypatch.setattr(time, "monotonic", lambda: now[0])
+    assert s.ensure_cache_available(old, 0) is False
+    now[0] += sched_mod._ADMIT_DEFER_TIMEOUT_S - 1
+    assert s.ensure_cache_available(new, 0) is False
+    assert s.get_unrecoverable_requests() == set()
+
+    now[0] += 2
+    # `old` is now past its budget; `new` is not.
+    assert s.ensure_cache_available(new, 0) is False
+    assert s.get_unrecoverable_requests() == set()
+    assert s.ensure_cache_available(old, 0) is False
+    assert s.get_unrecoverable_requests() == {"old"}
     s.shutdown()
 
 
@@ -250,7 +319,7 @@ def test_metadata_only_entry_admits_without_transfer(monkeypatch):
     s.shutdown()
 
 
-def test_orphan_not_ready_entry_falls_back_no_realloc(monkeypatch):
+def test_orphan_not_ready_entry_defers_no_realloc(monkeypatch):
     # Post-quarantine, post-tombstone-consumed orphan: a not-ready cache
     # entry exists for the mm_hash, but the hash is tracked in none of
     # _in_flight / _step_completed / _tombstones. Its blocks are held by a
@@ -274,25 +343,42 @@ def _spy_alloc(key, n):
     monkeypatch.setattr(s._cache, "alloc", _spy_alloc)
     req = _Request([_Feature("h1", 1)], params=_params("h1", 1))
 
-    # Must not raise; request falls back to local compute (admitted True),
-    # the orphan entry is untouched, and alloc is never called again.
-    assert s.ensure_cache_available(req, 0) is True
+    # Must not raise; the request is deferred while the DMA settles, the
+    # orphan entry is untouched, and alloc is never called again.
+    assert s.ensure_cache_available(req, 0) is False
     assert calls == []
     assert s._cache.get("h1") is entry
     assert not entry.ready
+    assert s.get_unrecoverable_requests() == set()
     s.shutdown()
 
 
-def test_alloc_failure_falls_back_to_local(monkeypatch):
+def test_full_pool_defers_then_fails_the_request(monkeypatch):
+    """A full pool is transient, so defer — but not forever.
+
+    Deferring without a budget would hold the request until the client gave
+    up, with no record of why.
+    """
+    import time as _time
+
     s = _consumer_sched(monkeypatch)
 
     # Force the cache to reject the allocation.
     monkeypatch.setattr(s._cache, "alloc", lambda key, n: None)
     req = _Request([_Feature("h1", 1)], params=_params("h1", 1))
 
-    # _start_xfer returns False -> request admitted for local recompute.
-    assert s.ensure_cache_available(req, 0) is True
+    t0 = _time.monotonic()
+    monkeypatch.setattr(_time, "monotonic", lambda: t0)
+    assert s.ensure_cache_available(req, 0) is False
     assert "h1" not in s._in_flight
+    assert s.get_unrecoverable_requests() == set()
+
+    # Still full once the budget has elapsed: fail rather than defer again.
+    monkeypatch.setattr(
+        _time, "monotonic", lambda: t0 + sched_mod._ADMIT_DEFER_TIMEOUT_S + 1
+    )
+    assert s.ensure_cache_available(req, 0) is False
+    assert s.get_unrecoverable_requests() == {"r1"}
     s.shutdown()
 
 

tests/v1/ec_connector/unit/test_scheduler_nixl_producer.py

@@ -111,3 +111,132 @@ def test_request_finished_skips_unallocated_entry(monkeypatch):
     # no cache entry to transfer (empty "metadata": no fields, no transfer).
     assert params == {"h1": {"metadata": {}}}
     s.shutdown()
+
+
+# ── announced encodings are held until read ──────────────────────────────────
+
+
+class _FakeProducerSession:
+    """Stands in for ProducerSession so build_connector_meta can run."""
+
+    def __init__(self):
+        self.served: list[str] = []
+
+    def poll_step(self):
+        pass
+
+    def take_served(self):
+        served, self.served = self.served, []
+        return served
+
+
+def _announcing_sched(monkeypatch, lease=30.0):
+    s = _sched_gate_off(monkeypatch)
+    s._nixl_enabled = True
+    s._peer_host, s._peer_port = "1.2.3.4", 5601
+    s._hidden_dim, s._element_size = 32, 2
+    s._announce_lease_s = lease
+    s._producer_session = _FakeProducerSession()
+    return s
+
+
+def _announce(s, mm_hash, length=2):
+    return s.request_finished(_Request([_Feature(mm_hash, length=length)]))
+
+
+def test_announced_encoding_is_not_evictable(monkeypatch):
+    """Announcing publishes an address the consumer uses on a later step.
+
+    By then the orchestrator has rewritten the media off the request, so an
+    eviction inside that window leaves the consumer nothing to fall back on.
+    """
+    s = _announcing_sched(monkeypatch)
+    entry = s._cache.alloc("h1", 2)
+    assert entry is not None
+    s._cache.mark_ready("h1")
+    assert entry.evictable
+
+    _delay, params = _announce(s, "h1")
+    assert "peer_host" in params["h1"]
+    assert not entry.evictable
+    s.shutdown()
+
+
+def test_hold_is_released_once_the_read_lands(monkeypatch):
+    s = _announcing_sched(monkeypatch)
+    entry = s._cache.alloc("h1", 2)
+    s._cache.mark_ready("h1")
+    _announce(s, "h1")
+    assert not entry.evictable
+
+    s._producer_session.served.append("h1")
+    s.build_connector_meta(scheduler_output=None)
+    assert entry.evictable
+    s.shutdown()
+
+
+def test_hold_lapses_when_no_consumer_ever_reads(monkeypatch):
+    """A consumer that never asks must not pin the pool forever."""
+    s = _announcing_sched(monkeypatch, lease=0.0)
+    entry = s._cache.alloc("h1", 2)
+    s._cache.mark_ready("h1")
+    _announce(s, "h1")
+    assert not entry.evictable
+
+    import time as _time
+
+    _time.sleep(0.01)
+    s.build_connector_meta(scheduler_output=None)
+    assert entry.evictable
+    s.shutdown()
+
+
+def test_hold_is_taken_when_a_late_save_lands(monkeypatch):
+    """An entry announced mid-save is pinned when it becomes evictable.
+
+    A not-ready entry cannot be evicted, so the hold has to start at
+    mark_ready rather than at the announcement.
+    """
+    from vllm.distributed.ec_transfer.ec_connector.cpu.common import (
+        ECCPUWorkerMetadata,
+    )
+    from vllm.v1.outputs import ECConnectorOutput
+
+    s = _announcing_sched(monkeypatch)
+    entry = s._cache.alloc("h1", 2)
+    assert entry is not None and not entry.ready
+    _announce(s, "h1")
+    assert s._announce["h1"].pending == 1
+
+    s.update_connector_output(
+        ECConnectorOutput(
+            ec_connector_worker_meta=ECCPUWorkerMetadata(completed_saves=["h1"])
+        )
+    )
+    assert entry.ready
+    assert not entry.evictable
+    assert not s._announce["h1"].pending
+    s.shutdown()
+
+
+def test_each_announcement_keeps_its_own_hold(monkeypatch):
+    """Two consumers told to read the same encoding each need it to survive.
+
+    Releasing on the first read would let the entry be evicted while the
+    second consumer, whose media has already been rewritten to a remote
+    reference, has not started its read.
+    """
+    s = _announcing_sched(monkeypatch)
+    entry = s._cache.alloc("h1", 2)
+    s._cache.mark_ready("h1")
+    _announce(s, "h1")
+    _announce(s, "h1")
+
+    s._producer_session.served.append("h1")
+    s.build_connector_meta(scheduler_output=None)
+    assert not entry.evictable
+
+    s._producer_session.served.append("h1")
+    s.build_connector_meta(scheduler_output=None)
+    assert entry.evictable
+    s.shutdown()

tests/v1/ec_connector/unit/test_session.py

@@ -451,14 +451,29 @@ def test_consumer_session_poll_not_ready_nack_goes_to_retryable():
     assert not results.completed
 
 
+def test_consumer_session_ack_timeout_goes_to_retryable():
+    """A late XferAck means a busy producer, not a broken one.
+
+    The producer answers XferReqs from its scheduler step, so the reply can
+    lag a whole encoder forward pass. Tombstoning on that admits the request
+    with no embedding, which is fatal once the media has been rewritten away.
+    """
+    s = _make_consumer_session()
+    s.start_xfer("h1", [0, 1], deadline=time.monotonic() - 1)
+    s.poll([], time.monotonic())
+    results = s.take_results()
+    assert "h1" in results.retryable
+    assert not results.tombstoned
+
+
 def test_consumer_session_take_results_clears_state():
     s = _make_consumer_session()
     s.start_xfer("h1", [0, 1], deadline=time.monotonic() - 1)
     s.poll([], time.monotonic())
     r1 = s.take_results()
-    assert "h1" in r1.tombstoned
+    assert "h1" in r1.retryable
     r2 = s.take_results()
-    assert not r2.tombstoned  # cleared after first take
+    assert not r2.retryable  # cleared after first take
 
 
 def test_consumer_session_on_peer_down_cancels_waiting_ack():

vllm/distributed/ec_transfer/ec_connector/base.py

@@ -261,6 +261,22 @@ def ensure_cache_available(
         """
         return True
 
+    def get_unrecoverable_requests(self) -> set[str]:
+        """Requests that can never be scheduled and must be failed.
+
+        A remote encoding has no local fallback: unlike KV, which is always
+        recomputable from the prompt, an encoding announced by a producer
+        exists only there, because the media never reached this instance.
+        A connector reports a request here once it is certain the encoding
+        will not arrive, and the scheduler finishes it with an error reason —
+        the same treatment an unrecoverable KV load gets — rather than
+        running the model without its encoder outputs.
+
+        Returns:
+            Request IDs to fail. The set is drained by the call.
+        """
+        return set()
+
     @abstractmethod
     def update_state_after_alloc(self, request: "Request", index: int):
         """

vllm/distributed/ec_transfer/ec_connector/cpu/connector.py

@@ -104,6 +104,10 @@ def ensure_cache_available(
             request, num_computed_tokens
         )
 
+    def get_unrecoverable_requests(self) -> set[str]:
+        assert self.connector_scheduler is not None
+        return self.connector_scheduler.get_unrecoverable_requests()
+
     def update_state_after_alloc(self, request: "Request", index: int) -> None:
         assert self.connector_scheduler is not None
         self.connector_scheduler.update_state_after_alloc(request, index)