[BugFix][TTS] R1.1-R1.4: turn silent async-chunk failures into visible ones (#6033)

Signed-off-by: Yueqian Lin <linyueqian@outlook.com>
Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>

tests/core/sched/test_omni_scheduler_mixin_shared.py

@@ -4,7 +4,10 @@ from types import SimpleNamespace
 import pytest
 from vllm.v1.engine import FinishReason
 
-from vllm_omni.core.sched.omni_scheduler_mixin import OmniSchedulerMixin
+from vllm_omni.core.sched.omni_scheduler_mixin import (
+    DEFAULT_INPUT_WAIT_TIMEOUT_S,
+    OmniSchedulerMixin,
+)
 from vllm_omni.core.sched.output import OmniChunkRecvHandle
 
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
@@ -22,13 +25,25 @@ def test_schedule_lifecycle_helpers_process_and_restore_both_input_paths():
     scheduler.requests = {"request": object()}
     scheduler._consume_pending_connector_output = lambda mode: calls.append(("consume", mode))
     scheduler._process_pending_input_timeouts = lambda: calls.append(("timeouts",))
+
+    def _collect_timed_out(timeout_s):
+        calls.append(("chunk-timeouts", timeout_s))
+        return set()
+
+    def _collect_failed_sends():
+        calls.append(("failed-sends",))
+        return {}
+
     scheduler.chunk_transfer_adapter = SimpleNamespace(
+        receives_chunks=True,
         process_pending_chunks=lambda waiting, running, scheduler_requests: calls.append(
             ("process", waiting, running, scheduler_requests)
         ),
         restore_queues=lambda waiting, running, scheduler_requests: calls.append(
             ("restore-chunks", waiting, running, scheduler_requests)
         ),
+        collect_timed_out_request_ids=_collect_timed_out,
+        collect_failed_send_request_ids=_collect_failed_sends,
     )
     scheduler.input_coordinator = SimpleNamespace(
         restore_queues=lambda waiting: calls.append(("restore-full", waiting))
@@ -41,6 +56,10 @@ def test_schedule_lifecycle_helpers_process_and_restore_both_input_paths():
         ("consume", "ar"),
         ("timeouts",),
         ("process", scheduler.waiting, scheduler.running, scheduler.requests),
+        # The chunk deadline runs after chunks are applied, so a chunk that
+        # arrived this cycle resets the clock before it is measured (R1.1).
+        ("chunk-timeouts", DEFAULT_INPUT_WAIT_TIMEOUT_S),
+        ("failed-sends",),
         ("restore-chunks", scheduler.waiting, scheduler.running, scheduler.requests),
         ("restore-full", scheduler.waiting),
     ]

tests/core/sched/test_omni_scheduler_mixin_timeouts.py

@@ -1,6 +1,10 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Unit coverage for _process_pending_input_timeouts.
+"""Unit coverage for the two stage-input timeout hooks.
+
+``_process_pending_input_timeouts`` covers the full-payload transport;
+``_process_pending_chunk_timeouts`` covers async-chunk. Both share
+``VLLM_OMNI_INPUT_WAIT_TIMEOUT_S``.
 
 Verifies that the mixin correctly *delegates* timed-out requests to the
 base scheduler's ``finish_requests`` API with ``RequestStatus.FINISHED_ERROR``.
@@ -12,6 +16,10 @@ this file only asserts the wiring from the mixin to that API.
 
 from __future__ import annotations
 
+import importlib
+import logging
+import math
+import os
 from types import SimpleNamespace
 
 import pytest
@@ -102,5 +110,248 @@ def test_process_pending_input_timeouts_disabled_when_timeout_zero(monkeypatch):
     assert scheduler.finish_calls == []
 
 
+class _FakeChunkAdapter:
+    """Stands in for OmniChunkTransferAdapter's timeout-collection surface."""
+
+    def __init__(self, timed_out_ids, *, receives_chunks=True):
+        self._timed_out_ids = set(timed_out_ids)
+        self.receives_chunks = receives_chunks
+        self.calls = []
+
+    def collect_timed_out_request_ids(self, timeout_s):
+        self.calls.append(timeout_s)
+        return set(self._timed_out_ids)
+
+
+class _FakeChunkScheduler(OmniSchedulerMixin):
+    def __init__(self, requests, adapter):
+        self.requests = requests
+        self.chunk_transfer_adapter = adapter
+        self.finish_calls = []
+
+    def finish_requests(self, req_ids, status):
+        self.finish_calls.append((set(req_ids), status))
+
+
+def test_process_pending_chunk_timeouts_delegates_to_finish_requests():
+    """A request stalled in WAITING_FOR_CHUNK is failed, not left parked.
+
+    This is the async-chunk half of the safety net (RFC #4855 R1.1); before it
+    existed, the chunk path had no deadline at all and issue #3833 could park a
+    request forever.
+    """
+    req_id = "stalled-stream"
+    adapter = _FakeChunkAdapter(timed_out_ids={req_id})
+    scheduler = _FakeChunkScheduler({req_id: SimpleNamespace(request_id=req_id)}, adapter)
+
+    scheduler._process_pending_chunk_timeouts()
+
+    assert len(adapter.calls) == 1
+    assert adapter.calls[0] > 0
+    assert len(scheduler.finish_calls) == 1
+    finished_ids, status = scheduler.finish_calls[0]
+    assert finished_ids == {req_id}
+    assert getattr(status, "name", str(status)).endswith("FINISHED_ERROR")
+
+
+def test_process_pending_chunk_timeouts_skips_already_freed_request():
+    adapter = _FakeChunkAdapter(timed_out_ids={"already-freed"})
+    scheduler = _FakeChunkScheduler(requests={}, adapter=adapter)
+
+    scheduler._process_pending_chunk_timeouts()
+
+    assert scheduler.finish_calls == []
+
+
+def test_process_pending_chunk_timeouts_noop_on_a_producer_stage():
+    """Stage 0 does not receive chunks, so it has nothing to expire."""
+    adapter = _FakeChunkAdapter(timed_out_ids={"r1"}, receives_chunks=False)
+    scheduler = _FakeChunkScheduler({"r1": SimpleNamespace(request_id="r1")}, adapter)
+
+    scheduler._process_pending_chunk_timeouts()
+
+    assert adapter.calls == []
+    assert scheduler.finish_calls == []
+
+
+def test_process_pending_chunk_timeouts_noop_without_adapter():
+    class _NoAdapter(OmniSchedulerMixin):
+        def __init__(self):
+            self.requests = {}
+            self.chunk_transfer_adapter = None
+            self.finish_calls = []
+
+        def finish_requests(self, req_ids, status):
+            self.finish_calls.append((set(req_ids), status))
+
+    scheduler = _NoAdapter()
+    scheduler._process_pending_chunk_timeouts()
+    assert scheduler.finish_calls == []
+
+
+def test_process_pending_chunk_timeouts_disabled_when_timeout_zero(monkeypatch):
+    """The shared knob disables both transports' nets together."""
+    from vllm_omni.core.sched import omni_scheduler_mixin
+
+    monkeypatch.setattr(omni_scheduler_mixin, "DEFAULT_INPUT_WAIT_TIMEOUT_S", 0.0)
+
+    adapter = _FakeChunkAdapter(timed_out_ids={"r1"})
+    scheduler = _FakeChunkScheduler({"r1": SimpleNamespace(request_id="r1")}, adapter)
+    scheduler._process_pending_chunk_timeouts()
+
+    assert adapter.calls == []
+    assert scheduler.finish_calls == []
+
+
+class _FakeSendFailureAdapter:
+    """Stands in for the adapter's drained send-failure map (R1.2)."""
+
+    def __init__(self, failures):
+        self._failures = dict(failures)
+        self.collect_calls = 0
+
+    def collect_failed_send_request_ids(self):
+        self.collect_calls += 1
+        failures, self._failures = self._failures, {}
+        return failures
+
+
+class _FakeSendFailureScheduler(OmniSchedulerMixin):
+    def __init__(self, requests, adapter):
+        self.requests = requests
+        self.chunk_transfer_adapter = adapter
+
+
+def test_log_failed_chunk_sends_reports_a_live_request(caplog):
+    adapter = _FakeSendFailureAdapter({"req-live": "OSError: No space left on device"})
+    scheduler = _FakeSendFailureScheduler({"req-live": SimpleNamespace(request_id="req-live")}, adapter)
+
+    with caplog.at_level(logging.ERROR):
+        scheduler._log_failed_chunk_sends()
+
+    assert "req-live" in caplog.text
+    assert "No space left on device" in caplog.text
+
+
+def test_log_failed_chunk_sends_still_reports_a_departed_request(caplog):
+    """``collect_*`` drains the map, so a failure for a request that already left
+    this scheduler is the one place a drop could still go unrecorded."""
+    adapter = _FakeSendFailureAdapter({"req-gone": "OSError: No space left on device"})
+    scheduler = _FakeSendFailureScheduler({}, adapter)
+
+    with caplog.at_level(logging.WARNING):
+        scheduler._log_failed_chunk_sends()
+
+    assert adapter.collect_calls == 1
+    assert "req-gone" in caplog.text
+    assert "No space left on device" in caplog.text
+    assert "finished or aborted" in caplog.text
+
+
+def test_log_failed_chunk_sends_noop_without_adapter():
+    scheduler = _FakeSendFailureScheduler({}, None)
+    scheduler._log_failed_chunk_sends()
+
+
+# --- R1.4: VLLM_OMNI_INPUT_WAIT_TIMEOUT_S is parsed at import time, so each of
+# these reloads the module under a different environment. One knob now gates the
+# deadline on both transports, which is why a silent 0 matters more than it did
+# when only the full-payload path used it.
+
+
+@pytest.fixture
+def reload_with(monkeypatch):
+    """Reload the mixin module under a given env value, then put it back.
+
+    ``importlib.reload`` re-executes into the live module dict, so whatever the
+    last test parsed is what every later test in the session sees -- ``-k
+    zero_warns`` alone would leave ``DEFAULT_INPUT_WAIT_TIMEOUT_S = 0.0`` and
+    silently disable both deadlines. The rejection cases are worse: the
+    ValueError fires *after* the rebind, so ``nan`` lingers in the module.
+    ``monkeypatch`` restores the environment but not the module, so reload once
+    more on the way out.
+    """
+    import vllm_omni.core.sched.omni_scheduler_mixin as mod
+
+    def _reload(raw):
+        if raw is None:
+            monkeypatch.delenv("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", raising=False)
+        else:
+            monkeypatch.setenv("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", raw)
+        return importlib.reload(mod)
+
+    yield _reload
+
+    monkeypatch.undo()
+    importlib.reload(mod)
+
+
+@pytest.mark.parametrize(
+    "raw,expected",
+    [
+        (None, 600.0),
+        ("30", 30.0),
+        ("0.5", 0.5),
+        ("0", 0.0),  # explicit opt-out is still honoured
+    ],
+)
+def test_timeout_accepts_valid_values(reload_with, raw, expected):
+    mod = reload_with(raw)
+    assert mod.DEFAULT_INPUT_WAIT_TIMEOUT_S == expected
+
+
+@pytest.mark.parametrize("raw", ["-1", "-600", "-0.5"])
+def test_negative_timeout_is_rejected(reload_with, raw):
+    """`-1` is a common "no limit" idiom; substituting 600 would give an operator
+    who meant "never time out" mysterious failures ten minutes in. Fail loudly."""
+    with pytest.raises(ValueError, match="negative"):
+        reload_with(raw)
+
+
+@pytest.mark.parametrize("raw", ["nan", "inf", "-inf", "1e999"])
+def test_non_finite_timeout_is_rejected(reload_with, raw):
+    """nan compares false against every deadline check and inf is unreachable;
+    both disable the net while looking like a number."""
+    with pytest.raises(ValueError, match="not finite"):
+        reload_with(raw)
+
+
+def test_zero_warns_and_names_both_transports(reload_with, caplog):
+    with caplog.at_level(logging.WARNING):
+        mod = reload_with("0")
+    assert mod.DEFAULT_INPUT_WAIT_TIMEOUT_S == 0.0
+    text = caplog.text
+    assert "DISABLED" in text
+    assert "WAITING_FOR_INPUT" in text and "WAITING_FOR_CHUNK" in text
+
+
+def test_unparsable_timeout_falls_back(reload_with):
+    mod = reload_with("not-a-number")
+    assert mod.DEFAULT_INPUT_WAIT_TIMEOUT_S == 600.0
+
+
+def test_a_rejected_value_does_not_outlive_its_test(monkeypatch):
+    """The reason ``reload_with`` reloads on the way out, done here by hand.
+
+    ``nan`` is bound before the ValueError fires, so the reload that rejects it
+    still leaves it in the module -- and nan compares false against every
+    deadline check, which is exactly the silent-disable this item exists to
+    stop. Restoring the environment is not enough; the module has to be reloaded
+    too.
+    """
+    import vllm_omni.core.sched.omni_scheduler_mixin as mod
+
+    ambient = float(os.environ.get("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", "600"))
+
+    monkeypatch.setenv("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", "nan")
+    with pytest.raises(ValueError, match="not finite"):
+        importlib.reload(mod)
+    assert math.isnan(mod.DEFAULT_INPUT_WAIT_TIMEOUT_S)
+
+    monkeypatch.undo()
+    importlib.reload(mod)
+    assert mod.DEFAULT_INPUT_WAIT_TIMEOUT_S == ambient
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])

tests/distributed/omni_connectors/test_chunk_transfer_adapter.py

@@ -1728,3 +1728,112 @@ def test_purge_is_noop_on_empty_deques(build_adapter):
     adapter.restore_queues(waiting_queue, running_queue, scheduler_requests={})
     assert running_queue == []
     assert waiting_queue == []
+
+
+# --------------------------------------------------------------------------- #
+#  Chunk-wait deadline (RFC #4855 R1.1, issue #3833)
+#
+#  Before this, the async-chunk path had no deadline of any kind, so a dropped
+#  terminal chunk or an upstream stage that died mid-stream parked the request
+#  in WAITING_FOR_CHUNK forever. None of these scenarios were covered.
+# --------------------------------------------------------------------------- #
+
+
+def _park_in_chunk_wait(adapter, request, *, waiting=True):
+    """Drive a request through one process_pending_chunks round into the wait."""
+    queue = DummyWaitingQueue([request]) if waiting else [request]
+    if waiting:
+        adapter.process_pending_chunks(queue, [])
+    else:
+        adapter.process_pending_chunks(DummyWaitingQueue(), queue)
+    return queue
+
+
+def test_chunk_wait_clock_starts_when_a_request_parks(build_adapter):
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    request = _req("r1", RequestStatus.WAITING)
+
+    _park_in_chunk_wait(adapter, request)
+
+    assert request.status == RequestStatus.WAITING_FOR_CHUNK
+    assert "r1" in adapter._waiting_since
+
+
+def test_chunk_wait_does_not_expire_before_the_deadline(build_adapter):
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    _park_in_chunk_wait(adapter, _req("r1", RequestStatus.WAITING))
+
+    assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == set()
+    assert "r1" in adapter._waiting_since
+
+
+def test_dropped_terminal_chunk_expires_the_request(build_adapter):
+    """The #3833 scenario: upstream stops sending and never marks the stream done."""
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    _park_in_chunk_wait(adapter, _req("r1", RequestStatus.WAITING))
+
+    adapter._waiting_since["r1"] -= 601.0
+
+    assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == {"r1"}
+    # Cleared, so a second sweep does not re-report the same request.
+    assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == set()
+
+
+def test_arriving_chunk_resets_the_clock(build_adapter):
+    """A slow but healthy stream must never expire: the deadline measures stall
+    time between chunks, not the lifetime of the stream."""
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    request = _req("r1", RequestStatus.WAITING)
+    queue = _park_in_chunk_wait(adapter, request)
+    adapter.restore_queues(queue, [], scheduler_requests={"r1": request})
+
+    # Age the wait to just under the deadline, then deliver a chunk.
+    adapter._waiting_since["r1"] -= 599.0
+    adapter._finished_load_reqs.add("r1")
+    adapter.process_pending_chunks(queue, [])
+
+    assert "r1" not in adapter._waiting_since
+    assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == set()
+
+
+def test_a_disabled_deadline_never_expires(build_adapter):
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    _park_in_chunk_wait(adapter, _req("r1", RequestStatus.WAITING))
+    adapter._waiting_since["r1"] -= 10_000.0
+
+    assert adapter.collect_timed_out_request_ids(timeout_s=0.0) == set()
+    assert adapter.collect_timed_out_request_ids(timeout_s=-1.0) == set()
+
+
+def test_finished_request_leaves_no_stale_timestamp(build_adapter):
+    """Otherwise a completed request would be reported as timed out later."""
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    request = _req("r1", RequestStatus.WAITING)
+    _park_in_chunk_wait(adapter, request)
+
+    adapter.finish_requests(["r1"], RequestStatus.FINISHED_STOPPED, {"r1": request})
+
+    assert "r1" not in adapter._waiting_since
+    assert adapter.collect_timed_out_request_ids(timeout_s=0.001) == set()
+
+
+def test_aborted_request_leaves_no_stale_timestamp(build_adapter):
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    _park_in_chunk_wait(adapter, _req("r1", RequestStatus.WAITING))
+
+    adapter.cleanup_receiver("r1")
+
+    assert "r1" not in adapter._waiting_since
+
+
+def test_expiry_is_per_request(build_adapter):
+    """One stalled stream must not take down its healthy neighbours."""
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar", max_num_seqs=4)
+    stalled = _req("stalled", RequestStatus.WAITING)
+    healthy = _req("healthy", RequestStatus.WAITING)
+    adapter.process_pending_chunks(DummyWaitingQueue([stalled, healthy]), [])
+
+    adapter._waiting_since["stalled"] -= 601.0
+
+    assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == {"stalled"}
+    assert "healthy" in adapter._waiting_since

tests/engine/test_orchestrator_error_handling.py

@@ -411,6 +411,51 @@ async def test_async_chunk_prewarm_failure_reports_failing_downstream_stage(orch
             orchestrator_fixture.thread.join(timeout=5)
 
 
+@pytest.mark.asyncio
+async def test_async_chunk_prewarm_without_prompt_token_ids_fails_only_that_request(
+    orchestrator_factory,
+) -> None:
+    """R1.3 of #4855: an async-chunk request whose stage-0 prompt carries no
+    token ids cannot prewarm anything, so it must fail with a client error
+    instead of stalling.
+
+    Non-fatal is the point: ``fatal=True`` reads as "the engine died" and the
+    entrypoints act on it (``OmniLLM.generate`` raises out of its batch loop,
+    the serving layer calls ``terminate_if_errored``). One client's embeds-only
+    prompt must not reach other requests.
+    """
+    stage0 = FakeStageClient(stage_type="llm", final_output=False)
+    stage1 = FakeStageClient(stage_type="llm", final_output=True)
+    orchestrator_fixture = orchestrator_factory([stage0, stage1], async_chunk=True)
+
+    try:
+        await _enqueue_add_request(
+            orchestrator_fixture,
+            request_id="req-embeds",
+            # No prompt_token_ids: what an embeds-only prompt looks like here.
+            prompt=SimpleNamespace(request_id="req-embeds"),
+            original_prompt={"prompt": "embeds only"},
+            sampling_params_list=[_sampling_params(), _sampling_params()],
+            final_stage_id=1,
+        )
+
+        error_msg = await _wait_for_error_message(orchestrator_fixture, request_id="req-embeds")
+        assert error_msg.fatal is False
+        assert error_msg.status_code == 400
+        assert error_msg.error_type == "BadRequestError"
+        assert error_msg.stage_id == 0
+        assert "prompt_token_ids" in error_msg.error
+
+        # Nothing was prewarmed downstream, and the request is gone.
+        assert stage1.add_request_calls == []
+        await _wait_for(lambda: "req-embeds" not in orchestrator_fixture.orchestrator.request_states)
+        assert orchestrator_fixture.thread.is_alive()
+    finally:
+        if orchestrator_fixture.thread.is_alive():
+            orchestrator_fixture.request_sync_q.put_nowait(ShutdownRequestMessage())
+            orchestrator_fixture.thread.join(timeout=5)
+
+
 # ───────── Direct unit tests for the fault-isolation helpers ─────────
 
 

tests/engine/test_orchestrator_stage_input_bridge.py

@@ -256,12 +256,13 @@ async def test_async_prewarm_skips_outgoing_only_stage() -> None:
         duplex_identity=SimpleNamespace(),
     )
 
-    await orchestrator._prewarm_async_chunk_stages(
+    prewarmed = await orchestrator._prewarm_async_chunk_stages(
         "req-prewarm",
         SimpleNamespace(prompt_token_ids=[1, 2], resumable=True),
         req_state,
     )
 
+    assert prewarmed is True
     assert stage1.submitted == []
     assert len(stage2.submitted) == 1
     assert 1 not in req_state.stage_submit_ts
@@ -277,6 +278,7 @@ async def test_async_prewarm_skips_outgoing_only_stage() -> None:
 @pytest.mark.asyncio
 async def test_duplex_prewarm_runs_after_first_stage0_submission() -> None:
     port, stage_pools, request_states, prewarm, submission = _duplex_stage_port_submission()
+    prewarm.return_value = True
 
     result = await port.submit(submission)
 
@@ -285,6 +287,30 @@ async def test_duplex_prewarm_runs_after_first_stage0_submission() -> None:
     prewarm.assert_awaited_once_with("req-duplex", ANY, request_states["req-duplex"])
 
 
+@pytest.mark.asyncio
+async def test_duplex_submit_bails_out_when_prewarm_failed_the_request() -> None:
+    """A failed prewarm has already aborted the request and popped its state.
+
+    Returning a success result here would hand the control plane a replica for a
+    request that no longer exists, and the trailing bookkeeping would re-register
+    a running counter the cleanup just released -- a leak that never decrements.
+    """
+    port, stage_pools, request_states, prewarm, submission = _duplex_stage_port_submission()
+    counter = MagicMock()
+    port._running_counter = counter
+    prewarm.return_value = False
+    request_state = request_states["req-duplex"]
+
+    with pytest.raises(RuntimeError, match="prewarm failed"):
+        await port.submit(submission)
+
+    prewarm.assert_awaited_once()
+    assert request_state.duplex_stage_fences == {}
+    assert request_state.stage_submit_ts == {}
+    assert request_state.running_counter_registered is False
+    counter.increment.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_async_route_forwards_to_outgoing_only_stage() -> None:
     orchestrator = object.__new__(Orchestrator)

vllm_omni/core/sched/omni_scheduler_mixin.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import math
 import os
 import time
 from collections.abc import Iterable
@@ -42,25 +43,56 @@ logger = init_logger(__name__)
 
 _STATS_INTERVAL_S = 1.0
 
-# Upper bound on how long a request may sit in full-payload-input wait
-# (the state ``OmniSchedulingCoordinator`` records via ``_waiting_since``)
-# before the scheduler force-fails it.  Defends against stuck consumer-side
-# requests when the producer drops a full-payload, send fails, or recv
-# never arrives.  Override per-deployment via
-# VLLM_OMNI_INPUT_WAIT_TIMEOUT_S; set <=0 to disable the safety net.
+# Upper bound on how long a request may wait for stage input before the
+# scheduler force-fails it.  Defends against stuck consumer-side requests when
+# the producer drops a payload, send fails, or recv never arrives.  Override
+# per-deployment via VLLM_OMNI_INPUT_WAIT_TIMEOUT_S; set exactly 0 to disable
+# the safety net. Negative and non-finite values are rejected at startup.
 #
-# Scope: this constant only covers the full-payload coordinator path
-# (``input_coordinator``).  The async-chunk path uses
-# ``chunk_transfer_adapter`` and is not affected by this constant.
+# Scope: both transports.  The full-payload path measures time parked in
+# WAITING_FOR_INPUT (``OmniSchedulingCoordinator._waiting_since``); the
+# async-chunk path measures time stalled in WAITING_FOR_CHUNK
+# (``OmniChunkTransferAdapter._waiting_since``, reset on each chunk arrival).
+# One knob, because the question it answers -- "how long may a request wait for
+# stage input?" -- does not depend on which transport carries it.
+_INPUT_WAIT_TIMEOUT_DEFAULT = 600.0
 _INPUT_WAIT_TIMEOUT_RAW = os.environ.get("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", "600")
 try:
     DEFAULT_INPUT_WAIT_TIMEOUT_S: float = float(_INPUT_WAIT_TIMEOUT_RAW)
 except ValueError:
     logger.warning(
-        "Invalid VLLM_OMNI_INPUT_WAIT_TIMEOUT_S=%r; falling back to 600 seconds.",
+        "Invalid VLLM_OMNI_INPUT_WAIT_TIMEOUT_S=%r; falling back to %s seconds.",
         _INPUT_WAIT_TIMEOUT_RAW,
+        _INPUT_WAIT_TIMEOUT_DEFAULT,
+    )
+    DEFAULT_INPUT_WAIT_TIMEOUT_S = _INPUT_WAIT_TIMEOUT_DEFAULT
+
+if not math.isfinite(DEFAULT_INPUT_WAIT_TIMEOUT_S):
+    # nan compares false against everything, so every deadline check silently
+    # passes; inf makes the deadline unreachable. Both disable the net while
+    # looking like a number, which is the failure this item exists to stop.
+    raise ValueError(
+        f"VLLM_OMNI_INPUT_WAIT_TIMEOUT_S={_INPUT_WAIT_TIMEOUT_RAW!r} is not finite. "
+        "Use a positive number of seconds, or exactly 0 to disable the deadline."
+    )
+if DEFAULT_INPUT_WAIT_TIMEOUT_S < 0:
+    # Fail at startup rather than substituting a default. `-1` is a common idiom
+    # for "no limit", so silently turning it into 600 would give an operator who
+    # meant "never time out" mysterious failures ten minutes in.
+    raise ValueError(
+        f"VLLM_OMNI_INPUT_WAIT_TIMEOUT_S={_INPUT_WAIT_TIMEOUT_RAW!r} is negative. "
+        "Use a positive number of seconds, or exactly 0 to disable the deadline."
+    )
+elif DEFAULT_INPUT_WAIT_TIMEOUT_S == 0:
+    # Zero stays a supported opt-out, but it is no longer silent: it disables
+    # the deadline on BOTH transports, so a request whose stage input never
+    # arrives waits forever. Say which nets just went away.
+    logger.warning(
+        "VLLM_OMNI_INPUT_WAIT_TIMEOUT_S=0: stage-input deadlines are DISABLED for "
+        "both the full-payload path (WAITING_FOR_INPUT) and the async-chunk path "
+        "(WAITING_FOR_CHUNK). A request whose input never arrives will wait "
+        "indefinitely instead of failing."
     )
-    DEFAULT_INPUT_WAIT_TIMEOUT_S = 600.0
 
 
 class OmniSchedulerMixin:
@@ -184,6 +216,8 @@ class OmniSchedulerMixin:
                 self.running,
                 scheduler_requests=self.requests,
             )
+            self._process_pending_chunk_timeouts()
+            self._log_failed_chunk_sends()
 
     def _restore_omni_wait_queues(self) -> None:
         """Restore requests temporarily parked by Omni input gates."""
@@ -211,9 +245,9 @@ class OmniSchedulerMixin:
         Disabled when ``DEFAULT_INPUT_WAIT_TIMEOUT_S`` is <= 0.
 
         Scope: only covers ``input_coordinator`` (full-payload path).
-        Async-chunk requests park in ``chunk_transfer_adapter`` instead
-        and are not handled here -- if a similar safety net is needed
-        for the chunk path, it belongs in the chunk adapter.
+        Async-chunk requests park in ``chunk_transfer_adapter`` instead and
+        are handled by ``_process_pending_chunk_timeouts`` below, which shares
+        this timeout.
         """
         if DEFAULT_INPUT_WAIT_TIMEOUT_S <= 0:
             return
@@ -234,6 +268,84 @@ class OmniSchedulerMixin:
         )
         self.finish_requests(present_ids, RequestStatus.FINISHED_ERROR)
 
+    def _log_failed_chunk_sends(self) -> None:
+        """Surface chunks the sender gave up on (R1.2 of #4855).
+
+        ``save_loop`` pops a task with ``popleft()`` and never re-queues it, and
+        ``connector.put`` can report failure without raising, so a dropped chunk
+        is a give-up rather than a retry. Until this change the only trace was a
+        warning that printed ``None`` for the request id, because the task dict
+        has no ``request_id`` key.
+
+        This only reports. Failing the request from here does not work: the
+        producer is not the ``final_output`` stage, so an abort synthesized by
+        ``finish_requests`` is not turned into a client-visible termination by
+        ``Orchestrator._route_output`` (see the ``final_output`` guards there).
+        Making a producer-side drop client-visible needs an orchestrator-side
+        path like ``_fail_request_dead_stage``; the consumer's R1.1 deadline is
+        what actually ends the request today.
+        """
+        adapter = getattr(self, "chunk_transfer_adapter", None)
+        if adapter is None:
+            return
+        failures = adapter.collect_failed_send_request_ids()
+        for request_id, reason in failures.items():
+            if request_id in self.requests:
+                logger.error(
+                    "[OmniScheduler] req=%s: chunk send gave up (%s); the consumer will "
+                    "fail on the stage-input deadline",
+                    request_id,
+                    reason,
+                )
+            else:
+                # Drained but no longer scheduled here: the request finished or
+                # was aborted between the failed send and this sweep. Still say
+                # so -- collect_* empties the map, so staying quiet would make
+                # this the one drop path in a change whose point is visibility.
+                logger.warning(
+                    "[OmniScheduler] req=%s: chunk send gave up (%s) after the request "
+                    "left this scheduler (finished or aborted)",
+                    request_id,
+                    reason,
+                )
+
+    def _process_pending_chunk_timeouts(self) -> None:
+        """Force-fail requests stalled in ``WAITING_FOR_CHUNK`` too long.
+
+        The async-chunk counterpart of ``_process_pending_input_timeouts``,
+        called right after ``chunk_transfer_adapter.process_pending_chunks``.
+        Until now the chunk path had no safety net at all: the scheduler-side
+        comment on the full-payload net said a similar guard "belongs in the
+        chunk adapter", while the worker mixin assumed the full-payload net
+        already covered it.  Only the first was true, so a dropped terminal
+        chunk or a producer that exhausted its send retries parked the request
+        forever (vllm-project/vllm-omni#3833).  Since ``async_chunk: true`` is
+        the default for the flagship TTS pipelines, that was the default
+        streaming path.
+
+        Shares ``VLLM_OMNI_INPUT_WAIT_TIMEOUT_S`` with the full-payload net:
+        one knob for "how long may a request wait for stage input", regardless
+        of which transport carries it.  Disabled when the value is <= 0.
+        """
+        if DEFAULT_INPUT_WAIT_TIMEOUT_S <= 0:
+            return
+        adapter = getattr(self, "chunk_transfer_adapter", None)
+        if adapter is None or not getattr(adapter, "receives_chunks", False):
+            return
+        timed_out_ids = adapter.collect_timed_out_request_ids(timeout_s=DEFAULT_INPUT_WAIT_TIMEOUT_S)
+        if not timed_out_ids:
+            return
+        present_ids = {req_id for req_id in timed_out_ids if req_id in self.requests}
+        if not present_ids:
+            return
+        logger.warning(
+            "Marking %d request(s) as FINISHED_ERROR after stalling > %.0fs waiting for a chunk: %s",
+            len(present_ids),
+            DEFAULT_INPUT_WAIT_TIMEOUT_S,
+            sorted(present_ids),
+        )
+        self.finish_requests(present_ids, RequestStatus.FINISHED_ERROR)
+
     def _capture_omni_connector_output(self, model_runner_output: Any) -> None:
         """Stash the model runner's omni_connector_output for next schedule().
 

vllm_omni/distributed/omni_connectors/transfer_adapter/base.py

@@ -35,6 +35,8 @@ class OmniTransferAdapterBase:
         self.stop_event = threading.Event()
         self._recv_cond = threading.Condition()
         self._save_cond = threading.Condition()
+        self._send_failures: dict[str, str] = {}
+        self._send_failure_lock = threading.Lock()
 
         self.recv_thread = threading.Thread(target=self.recv_loop, daemon=True)
         self.recv_thread.start()
@@ -82,6 +84,24 @@ class OmniTransferAdapterBase:
                 elif not any_success and not self.stop_event.is_set():
                     self._recv_cond.wait(timeout=0.001)
 
+    def record_send_failure(self, request_id: str | None, reason: str) -> None:
+        """Note that a chunk for *request_id* will never be delivered.
+
+        Called from the sender thread; drained by the scheduler thread, hence
+        the lock. ``_waiting_since`` needs none because only the scheduler
+        touches it.
+        """
+        if not request_id:
+            return
+        with self._send_failure_lock:
+            self._send_failures.setdefault(request_id, reason)
+
+    def collect_failed_send_request_ids(self) -> dict[str, str]:
+        """Drain and return the requests whose send gave up."""
+        with self._send_failure_lock:
+            failures, self._send_failures = self._send_failures, {}
+        return failures
+
     def save_loop(self):
         """Loop to send outgoing data."""
         while not self.stop_event.is_set():
@@ -90,7 +110,18 @@ class OmniTransferAdapterBase:
                 try:
                     self._send_single_request(task)
                 except Exception as e:
-                    logger.warning(f"Error saving data for {task.get('request_id')}: {e}")
+                    # R1.2 of #4855. The task was popleft()-ed and is not re-queued,
+                    # so this is a give-up, not a retry: the consumer will wait for a
+                    # chunk that is never coming. Record it so the scheduler can fail
+                    # the request now instead of leaving it to the wait deadline.
+                    # The task dict carries the Request object, not a bare id --
+                    # `task.get("request_id")` is always None (it was that way in the
+                    # original warning too, which logged "None" for every failure).
+                    # Use the scheduler-side id: that is what `self.requests` is keyed
+                    # by, and what `finish_requests` expects.
+                    failed = getattr(task.get("request"), "request_id", None)
+                    logger.error("Send gave up for %s: %s", failed, e)
+                    self.record_send_failure(failed, f"{type(e).__name__}: {e}")
 
             with self._save_cond:
                 if not self._pending_save_reqs and not self.stop_event.is_set():

vllm_omni/distributed/omni_connectors/transfer_adapter/chunk_transfer_adapter.py

@@ -2,6 +2,7 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
 import importlib
+import time
 from collections import defaultdict, deque
 from collections.abc import Callable, Mapping
 from typing import Any
@@ -94,6 +95,12 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
         self._held_non_active: deque[Any] = deque()
         self.requests_num_chunks_sent: dict[str, int] = defaultdict(int)
         self._pending_streaming_prefills: dict[str, dict] = {}
+        # Monotonic timestamp of when each request last began waiting for a
+        # chunk, refreshed every time one arrives.  Read by
+        # collect_timed_out_request_ids() so a stream that stops advancing
+        # becomes a client-visible error instead of parking forever.  Mirrors
+        # OmniSchedulingCoordinator._waiting_since on the full-payload path.
+        self._waiting_since: dict[str, float] = {}
 
     @staticmethod
     def _is_truthy_scalar(value: Any) -> bool:
@@ -383,6 +390,24 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
             # has been sent successfully. This avoids cleanup->save races.
             if is_payload_finished:
                 self.cleanup(request.request_id, external_req_id)
+        else:
+            # R1.2 of #4855. connector.put returning False is a silent drop: no
+            # exception, no retry, and the caller's `if success:` block simply
+            # does not run. /dev/shm exhaustion in SharedMemoryConnector.put is
+            # one way to get here. Record it so the request fails now rather
+            # than parking in WAITING_FOR_CHUNK until the deadline.
+            logger.error(
+                "Chunk send failed for %s (stage %s -> %s); giving up on this chunk",
+                external_req_id,
+                stage_id,
+                next_stage_id,
+            )
+            # Key on the scheduler-side id. `external_req_id` is the user-facing id
+            # (InputProcessor renames request_id to an internal UUID and keeps the
+            # original in external_req_id, see async_omni_engine.py), while
+            # `self.requests` -- and therefore `finish_requests` -- is keyed by the
+            # internal one.
+            self.record_send_failure(request.request_id, "connector.put reported failure")
 
         if is_segment_finished:
             self.code_prompt_token_ids.pop(external_req_id, None)
@@ -438,6 +463,7 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
 
         self._cancelled_load_reqs.add(request_id)
         self._finished_load_reqs.discard(request_id)
+        self._waiting_since.pop(request_id, None)
 
     @staticmethod
     def _discard_from_chunk_deque(deque_list: deque[Any], request_id: str) -> None:
@@ -579,6 +605,42 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
         self._active_streams[request_id] = request
         return True
 
+    def collect_timed_out_request_ids(self, timeout_s: float) -> set[str]:
+        """Return IDs whose chunk wait has exceeded *timeout_s*.
+
+        The async-chunk path had no deadline of any kind: a request parks in
+        ``WAITING_FOR_CHUNK`` and the receiver re-queues it on every failed
+        poll (``transfer_adapter/base.py``) with no attempt counter, so a
+        producer that crashed, gave up after its send retries, or simply never
+        emitted a terminal chunk left the request waiting forever
+        (vllm-project/vllm-omni#3833).  The full-payload path has had a net
+        since ``OmniSchedulingCoordinator.collect_timed_out_request_ids``; this
+        is its async-chunk counterpart, and both are driven by the same
+        ``VLLM_OMNI_INPUT_WAIT_TIMEOUT_S``.
+
+        The clock measures *stall* time, not stream lifetime: it starts when a
+        request begins waiting for a chunk and resets each time one arrives, so
+        a long but healthy stream is never failed.
+
+        Clears ``_waiting_since`` for the expired IDs.  The caller marks them
+        ``FINISHED_ERROR`` via the scheduler's ``finish_requests``, which routes
+        back through this adapter's own ``finish_requests`` and releases the
+        rest of the per-request state.
+        """
+        if timeout_s <= 0 or not self._waiting_since:
+            return set()
+        now = time.monotonic()
+        timed_out_ids = {req_id for req_id, started in self._waiting_since.items() if now - started > timeout_s}
+        for req_id in timed_out_ids:
+            self._waiting_since.pop(req_id, None)
+            logger.warning(
+                "[Stage-%s] Request %s timed out waiting for a chunk (stalled > %.0fs)",
+                self.connector.stage_id,
+                req_id,
+                timeout_s,
+            )
+        return timed_out_ids
+
     @property
     def num_running_waiting_for_chunk(self) -> int:
         """Count running requests temporarily removed while awaiting a chunk."""
@@ -624,11 +686,15 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
                 # Requests that waiting for chunk
                 self.load_async(request)
                 request.status = RequestStatus.WAITING_FOR_CHUNK
+                self._waiting_since.setdefault(request.request_id, time.monotonic())
             else:
                 if request.request_id in finished_load_reqs:
                     request.status = target_status
                     finished_load_reqs.remove(request.request_id)
                     self.requests_with_ready_chunks.add(request.request_id)
+                    # A chunk landed: restart the clock for the next one, so the
+                    # deadline measures stall time rather than total stream time.
+                    self._waiting_since.pop(request.request_id, None)
                     continue
             queue.remove(request)
             self.requests_origin_status[request.request_id] = target_status
@@ -767,11 +833,15 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
                 # Requests that waiting for chunk
                 self.load_async(request)
                 request.status = RequestStatus.WAITING_FOR_CHUNK
+                self._waiting_since.setdefault(request.request_id, time.monotonic())
             else:
                 if request.request_id in finished_load_reqs:
                     request.status = target_status
                     finished_load_reqs.remove(request.request_id)
                     self.requests_with_ready_chunks.add(request.request_id)
+                    # A chunk landed: restart the clock for the next one, so the
+                    # deadline measures stall time rather than total stream time.
+                    self._waiting_since.pop(request.request_id, None)
                     continue
             queue.remove(request)
             self.requests_origin_status[request.request_id] = target_status
@@ -826,5 +896,6 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
             self.upstream_exhausted_requests.discard(req_id)
             self._finished_load_reqs.discard(req_id)
             self._cancelled_load_reqs.add(req_id)
+            self._waiting_since.pop(req_id, None)
 
         return []

vllm_omni/engine/orchestrator.py

@@ -15,6 +15,7 @@ import asyncio
 import time as _time
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
+from http import HTTPStatus
 from typing import TYPE_CHECKING, Any
 
 import janus
@@ -51,6 +52,7 @@ from vllm_omni.engine.messages import (
 from vllm_omni.engine.orchestrator_monitor import create_orch_monitor, replica_key
 from vllm_omni.engine.serialization import serialize_additional_information
 from vllm_omni.engine.stage_pool import StagePool, StageUnavailableError
+from vllm_omni.errors import DEFAULT_CLIENT_ERROR_TYPE
 from vllm_omni.metrics.prometheus import OmniRequestCounter
 from vllm_omni.metrics.stat_logger import OmniPrometheusStatLogger
 from vllm_omni.outputs import OmniRequestOutput
@@ -221,7 +223,7 @@ class _OrchestratorDuplexStagePort:
         async_chunk: bool,
         prewarm_async_chunk_stages: Callable[
             [str, Any, OrchestratorRequestState],
-            Awaitable[None],
+            Awaitable[bool],
         ],
     ) -> None:
         self._stage_pools = stage_pools
@@ -307,11 +309,23 @@ class _OrchestratorDuplexStagePort:
         else:
             replica_id = await pool.submit_initial(context.request_id, request_state, request, prompt_text=None)
             if self._async_chunk and context.stage_id == 0:
-                await self._prewarm_async_chunk_stages(
+                prewarmed = await self._prewarm_async_chunk_stages(
                     context.request_id,
                     request,
                     request_state,
                 )
+                if not prewarmed:
+                    # The prewarm already failed the request, aborted it, and
+                    # popped its state. Falling through would write fences and
+                    # a submit timestamp onto that orphaned object and
+                    # re-increment the running counter the cleanup just
+                    # released, then hand the control plane a success result
+                    # for a request that no longer exists. Raise instead:
+                    # ``DuplexControlPlane.handle_append`` turns this into an
+                    # error result for the append.
+                    raise RuntimeError(
+                        f"async-chunk prewarm failed for duplex request {context.request_id}; the request was aborted"
+                    )
         request_state.duplex_stage_fences[context.stage_id] = context.fence
         request_state.stage_submit_ts[context.stage_id] = _time.time()
         if not request_state.running_counter_registered and self._running_counter is not None:
@@ -2217,18 +2231,54 @@ class Orchestrator:
         request_id: str,
         stage0_request: Any,
         req_state: OrchestratorRequestState,
-    ) -> None:
-        """Pre-submit downstream stages for async-chunk mode."""
+    ) -> bool:
+        """Pre-submit downstream stages for async-chunk mode.
+
+        Returns False when the request was failed and cleaned up in here, so a
+        caller still holding ``req_state`` stops instead of recording state on
+        an object the cleanup already popped from ``request_states``.
+        """
         if req_state.final_stage_id <= 0:
-            return
+            return True
 
         prompt_token_ids = getattr(stage0_request, "prompt_token_ids", None)
         if prompt_token_ids is None:
-            logger.warning(
-                "[Orchestrator] async_chunk prewarm skipped for req=%s: stage0 prompt_token_ids missing",
+            # R1.3 of #4855. Skipping the prewarm leaves every downstream stage
+            # unsubmitted, so the request produces no output and no error -- it
+            # just stops. An embeds-only prompt in async-chunk mode is a config
+            # mistake, and the client should hear about it now rather than wait
+            # for the stage-input deadline to notice nothing ever arrived.
+            #
+            # Non-fatal, with a 4xx: `fatal=True` means "the engine is dead", and
+            # the entrypoints act on it -- `OmniBase._handle_output_message`
+            # raises `OmniEngineDeadError` out of `OmniLLM.generate`'s batch loop,
+            # failing every other request in that batch, and the serving layer
+            # calls `terminate_if_errored`. One client's bad prompt must not do
+            # either. This is the same shape `_handle_stage_error` uses for a
+            # request-scoped client error.
+            logger.error(
+                "[Orchestrator] req=%s: async_chunk prewarm needs stage0 prompt_token_ids "
+                "and none were provided; failing the request",
                 request_id,
             )
-            return
+            await self.output_async_queue.put(
+                ErrorMessage(
+                    error=(
+                        "async_chunk requires prompt_token_ids on the stage-0 request; "
+                        "an embeds-only prompt cannot prewarm downstream stages"
+                    ),
+                    status_code=HTTPStatus.BAD_REQUEST.value,
+                    error_type=DEFAULT_CLIENT_ERROR_TYPE,
+                    request_id=request_id,
+                    stage_id=0,
+                )
+            )
+            await self._cleanup_request_ids(
+                [request_id, *self._cfg_tracker.cleanup_parent(request_id)],
+                abort=True,
+                close_duplex_sessions=True,
+            )
+            return False
 
         for next_stage_id in range(1, req_state.final_stage_id + 1):
             next_pool = self.stage_pools[next_stage_id]
@@ -2303,7 +2353,7 @@ class Orchestrator:
             # Attribute the failure to the stage that failed, not stage 0; the
             # request is already cleaned up, so stop prewarming.
             if not submitted:
-                return
+                return False
 
             # The guard swallows submit_initial's return value; the pool binding
             # it recorded carries the same replica.
@@ -2329,6 +2379,8 @@ class Orchestrator:
                 tx_ms=_tx_ms,
             )
 
+        return True
+
     def _build_kv_sender_info(
         self,
         sender_stage_ids: list[int],

vllm_omni/worker/omni_connector_model_runner_mixin.py

@@ -406,11 +406,13 @@ class OmniConnectorModelRunnerMixin:
         # thread by one execute_model() cycle, especially when the request was
         # added after the current scheduler_output snapshot.
         #
-        # Orphaned pending recv entries (e.g. from upstream stage crash)
-        # are handled by OmniSchedulingCoordinator.collect_timed_out_request_ids()
-        # which detects wait-time violations.  The scheduler then removes the
-        # request from its queues, sets FINISHED_ERROR, and calls _free_request()
-        # which ultimately triggers cleanup_finished_request() here.
+        # Orphaned pending recv entries (e.g. from upstream stage crash) are
+        # handled by collect_timed_out_request_ids() -- on
+        # OmniSchedulingCoordinator for full-payload requests, and on
+        # OmniChunkTransferAdapter for async-chunk ones -- which detect
+        # wait-time violations.  The scheduler then removes the request from
+        # its queues, sets FINISHED_ERROR, and calls _free_request() which
+        # ultimately triggers cleanup_finished_request() here.
         for attr_name in (
             "_request_ids_mapping",
             "_get_req_chunk",