[Security] Fix P2P ZMQ socket exhaustion via unbounded peer sessions (#51504)

Signed-off-by: jperezde <jperezde@redhat.com>
Signed-off-by: Juan Pérez de Algaba <jperezde@redhat.com>
Co-authored-by: Robert Shaw <114415538+robertgshaw2-redhat@users.noreply.github.com>
Co-authored-by: Itay Etelis <92247226+Etelis@users.noreply.github.com>

docs/usage/security.md

@@ -303,6 +303,25 @@ Only deployments whose API clients are trusted should start the server with
 `--trust-request-mm-kwargs` to restore per-request overrides. Do not enable
 this option on an endpoint exposed to untrusted clients.
 
+### P2P KV-Offload Peer Session Limits
+
+Deployments that enable the P2P secondary tier (`OffloadingConnector` with `TieringOffloadingSpec` and a `p2p` secondary tier) accept peer addresses from `kv_transfer_params` in API requests. Without limits, an attacker with API access could submit many unique non-listening peer addresses, each consuming ZeroMQ sockets until the context quota is exhausted, crashing the EngineCore.
+
+vLLM enforces several configurable limits to prevent this:
+
+| Environment Variable | Default | Description |
+| --- | --- | --- |
+| `VLLM_P2P_MAX_PEERS` | `128` | Maximum concurrent P2P peer sessions. New peers (both outbound and inbound) are rejected when this limit is reached. |
+| `VLLM_P2P_HANDSHAKE_TIMEOUT_S` | `30` | Seconds to wait for a newly connected peer to complete the P2P handshake. Sessions that remain connected but not ready after this deadline are torn down. |
+| `VLLM_P2P_IDLE_TIMEOUT_S` | `300` | Seconds a P2P session may remain idle (no pending work, no messages) before it is eligible for eviction. Idle eviction runs only when remaining peer slots are within 8 of `VLLM_P2P_MAX_PEERS`. Set to `0` to disable idle eviction. |
+
+Connections that fail (ZMQ socket errors, handshake failures) are handled gracefully — the request falls back to local prefill instead of crashing the engine.
+
+**Recommendations:**
+
+- **Public-facing deployments with P2P enabled:** Lower `VLLM_P2P_MAX_PEERS` to the expected number of legitimate peers (e.g., `16` or `32`).
+- **Sensitive deployments:** Consider deploying behind a reverse proxy that strips or validates `kv_transfer_params` from API requests to prevent client-supplied peer addresses entirely.
+
 ## Tool Server and MCP Security
 
 vLLM supports connecting to external tool servers via the `--tool-server` argument. This enables models to call tools through the Responses API (`/v1/responses`). Tool server support works with all models — it is not limited to specific model architectures.

tests/test_envs.py

@@ -60,6 +60,24 @@ def test_p2p_side_channel_defaults_and_override(monkeypatch: pytest.MonkeyPatch)
     assert envs.VLLM_P2P_SIDE_CHANNEL_PORT == 5799
 
 
+def test_p2p_resource_limit_defaults_and_override(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    monkeypatch.delenv("VLLM_P2P_MAX_PEERS", raising=False)
+    monkeypatch.delenv("VLLM_P2P_HANDSHAKE_TIMEOUT_S", raising=False)
+    monkeypatch.delenv("VLLM_P2P_IDLE_TIMEOUT_S", raising=False)
+    assert envs.VLLM_P2P_MAX_PEERS == 128
+    assert envs.VLLM_P2P_HANDSHAKE_TIMEOUT_S == 30
+    assert envs.VLLM_P2P_IDLE_TIMEOUT_S == 300
+
+    monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "64")
+    monkeypatch.setenv("VLLM_P2P_HANDSHAKE_TIMEOUT_S", "15")
+    monkeypatch.setenv("VLLM_P2P_IDLE_TIMEOUT_S", "60")
+    assert envs.VLLM_P2P_MAX_PEERS == 64
+    assert envs.VLLM_P2P_HANDSHAKE_TIMEOUT_S == 15
+    assert envs.VLLM_P2P_IDLE_TIMEOUT_S == 60
+
+
 def test_getattr_with_cache(monkeypatch: pytest.MonkeyPatch):
     monkeypatch.setenv("VLLM_HOST_IP", "1.1.1.1")
     monkeypatch.setenv("VLLM_PORT", "1234")

tests/v1/kv_offload/tiering/p2p/test_manager.py

@@ -647,6 +647,7 @@ class _FakeSession:
         close_req_ids: list[str] | None = None,
         close_stores: list[int] | None = None,
         close_failed_serves: list[ReqContext] | None = None,
+        last_activity_at: float | None = None,
     ) -> None:
         self.peer_id = peer_id
         self.alive = alive
@@ -663,6 +664,9 @@ class _FakeSession:
         self.stores_added: list[tuple[str, list, object, int]] = []
         self.attached: list[object] = []
         self.finishes: list[str] = []
+        self.last_activity_at: float = (
+            last_activity_at if last_activity_at is not None else time.monotonic()
+        )
         # Mirror P2PSession._server._inflight (transfer_id → handle) and
         # P2PSession._client.has_active_loads for the shutdown-drain and
         # drain_jobs paths. Tests populate _server._inflight when needed.
@@ -1844,3 +1848,272 @@ class TestBindHostPortDefaults:
         mgr_b = self._construct(monkeypatch, host="localhost", port=5710)
         assert mgr_a._local_id == mgr_b._local_id == "localhost:5710"
         assert mgr_a._nixl_agent_name != mgr_b._nixl_agent_name
+
+
+# ---------------------------------------------------------------------------
+# Tests for session capacity limit (VLLM_P2P_MAX_PEERS)
+# ---------------------------------------------------------------------------
+
+
+class TestSessionCapLimit:
+    """_get_or_create_session and _accept_new_peers reject new peers when
+    the session count reaches VLLM_P2P_MAX_PEERS."""
+
+    def _make_with_control(self):
+        mgr = _make_manager()
+        created = []
+
+        class FakeControl:
+            def connect(self, peer_id):
+                conn = _RecordingConn(peer_id)
+                created.append(conn)
+                return conn
+
+            def poll(self):
+                return []
+
+        class FakeData:
+            block_len = 4096
+            base_addr = 0x1000
+            num_blocks = 16
+            config_fingerprint = ""
+
+            def get_agent_metadata(self):
+                return b"meta"
+
+            def add_remote_peer(self, *a, **k):
+                pass
+
+            def remove_remote_peer(self, pid):
+                pass
+
+        mgr._control = FakeControl()  # type: ignore[assignment]
+        mgr._data = FakeData()  # type: ignore[assignment]
+        return mgr, created
+
+    def test_outbound_rejected_at_capacity(self, monkeypatch):
+        monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "2")
+        import vllm.envs
+
+        vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None)
+
+        mgr, created = self._make_with_control()
+        mgr._sessions["peer-a:1"] = _FakeSession(peer_id="peer-a:1")  # type: ignore[assignment]
+        mgr._sessions["peer-b:2"] = _FakeSession(peer_id="peer-b:2")  # type: ignore[assignment]
+
+        result = mgr._get_or_create_session("peer-c:3")
+        assert result is None
+        assert "peer-c:3" not in mgr._sessions
+        assert len(created) == 0
+
+    def test_on_new_request_records_failure_on_cap(self, monkeypatch):
+        monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "1")
+        import vllm.envs
+
+        vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None)
+
+        mgr, _ = self._make_with_control()
+        mgr._sessions["existing:1"] = _FakeSession(peer_id="existing:1")  # type: ignore[assignment]
+
+        ctx = _req_context(
+            kv_params=_remote_prefiller_kv_params(
+                remote_host="10.0.0.2", remote_port=9000, kv_request_id="req-cap"
+            )
+        )
+        mgr.on_new_request(ctx)
+        assert "req-cap" in mgr._failed_req_ids
+
+    def test_inbound_rejected_at_capacity(self, monkeypatch):
+        monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "1")
+        import vllm.envs
+
+        vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None)
+
+        mgr, _ = self._make_with_control()
+        mgr._sessions["existing:1"] = _FakeSession(peer_id="existing:1")  # type: ignore[assignment]
+
+        conn = _RecordingConn("new-peer:2")
+        mgr._accept_new_peers([conn])
+        assert conn.close_calls == 1
+        assert "new-peer:2" not in mgr._sessions
+
+    def test_existing_session_reused_below_cap(self, monkeypatch):
+        monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "1")
+        import vllm.envs
+
+        vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None)
+
+        mgr, _ = self._make_with_control()
+        existing = _FakeSession(peer_id="existing:1")
+        mgr._sessions["existing:1"] = existing  # type: ignore[assignment]
+
+        result = mgr._get_or_create_session("existing:1")
+        assert result is existing
+
+
+# ---------------------------------------------------------------------------
+# Tests for graceful error boundary in _get_or_create_session
+# ---------------------------------------------------------------------------
+
+
+class TestGetOrCreateSessionErrorBoundary:
+    def test_connect_failure_returns_none(self):
+        mgr = _make_manager()
+
+        class FailingControl:
+            def connect(self, peer_id):
+                raise RuntimeError("socket exhausted")
+
+            def poll(self):
+                return []
+
+        mgr._control = FailingControl()  # type: ignore[assignment]
+        result = mgr._get_or_create_session("bad:1")
+        assert result is None
+        assert "bad:1" not in mgr._sessions
+
+    def test_session_construction_failure_closes_conn(self):
+        """connect() succeeding but P2PSession construction failing must
+        close the conn so it is not left registered without a session."""
+        mgr = _make_manager()
+        created: list[_RecordingConn] = []
+
+        class RaisingConn(_RecordingConn):
+            def send(self, msg):
+                raise RuntimeError("send failed during connect")
+
+        class Control:
+            def connect(self, peer_id):
+                conn = RaisingConn(peer_id)
+                created.append(conn)
+                return conn
+
+            def poll(self):
+                return []
+
+        class FakeData:
+            block_len = 4096
+            base_addr = 0x1000
+            num_blocks = 16
+            config_fingerprint = ""
+
+            def get_agent_metadata(self):
+                return b"meta"
+
+        mgr._control = Control()  # type: ignore[assignment]
+        mgr._data = FakeData()  # type: ignore[assignment]
+        result = mgr._get_or_create_session("bad:2")
+        assert result is None
+        assert "bad:2" not in mgr._sessions
+        assert len(created) == 1
+        assert created[0].close_calls == 1
+
+
+# ---------------------------------------------------------------------------
+# Tests for idle session eviction
+# ---------------------------------------------------------------------------
+
+
+class TestIdleSessionEviction:
+    def _make_with_fake_data(self):
+        mgr = _make_manager()
+
+        class FakeControl:
+            def poll(self):
+                return []
+
+        class FakeData:
+            def remove_remote_peer(self, pid):
+                pass
+
+        mgr._control = FakeControl()  # type: ignore[assignment]
+        mgr._data = FakeData()  # type: ignore[assignment]
+        return mgr
+
+    def test_idle_session_evicted(self, monkeypatch):
+        monkeypatch.setenv("VLLM_P2P_IDLE_TIMEOUT_S", "10")
+        monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "1")
+        import vllm.envs
+
+        vllm.envs.__dict__.pop("VLLM_P2P_IDLE_TIMEOUT_S", None)
+        vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None)
+
+        mgr = self._make_with_fake_data()
+        idle_session = _FakeSession(
+            peer_id="idle:1",
+            last_activity_at=time.monotonic() - 20,
+        )
+        mgr._sessions["idle:1"] = idle_session  # type: ignore[assignment]
+
+        mgr._reap_idle_sessions()
+        assert "idle:1" not in mgr._sessions
+
+    def test_idle_session_kept_below_headroom(self, monkeypatch):
+        monkeypatch.setenv("VLLM_P2P_IDLE_TIMEOUT_S", "10")
+        import vllm.envs
+
+        vllm.envs.__dict__.pop("VLLM_P2P_IDLE_TIMEOUT_S", None)
+        vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None)
+
+        mgr = self._make_with_fake_data()
+        idle_session = _FakeSession(
+            peer_id="idle:1",
+            last_activity_at=time.monotonic() - 20,
+        )
+        mgr._sessions["idle:1"] = idle_session  # type: ignore[assignment]
+
+        mgr._reap_idle_sessions()
+        assert "idle:1" in mgr._sessions
+
+    def test_active_session_kept(self, monkeypatch):
+        monkeypatch.setenv("VLLM_P2P_IDLE_TIMEOUT_S", "10")
+        monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "1")
+        import vllm.envs
+
+        vllm.envs.__dict__.pop("VLLM_P2P_IDLE_TIMEOUT_S", None)
+        vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None)
+
+        mgr = self._make_with_fake_data()
+        active = _FakeSession(
+            peer_id="active:1",
+            last_activity_at=time.monotonic(),
+        )
+        mgr._sessions["active:1"] = active  # type: ignore[assignment]
+
+        mgr._reap_idle_sessions()
+        assert "active:1" in mgr._sessions
+
+    def test_session_with_pending_work_not_evicted(self, monkeypatch):
+        monkeypatch.setenv("VLLM_P2P_IDLE_TIMEOUT_S", "1")
+        monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "1")
+        import vllm.envs
+
+        vllm.envs.__dict__.pop("VLLM_P2P_IDLE_TIMEOUT_S", None)
+        vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None)
+
+        mgr = self._make_with_fake_data()
+        busy = _FakeSession(
+            peer_id="busy:1",
+            last_activity_at=time.monotonic() - 100,
+        )
+        busy._server._inflight[1] = object()
+        mgr._sessions["busy:1"] = busy  # type: ignore[assignment]
+
+        mgr._reap_idle_sessions()
+        assert "busy:1" in mgr._sessions
+
+    def test_disabled_when_zero(self, monkeypatch):
+        monkeypatch.setenv("VLLM_P2P_IDLE_TIMEOUT_S", "0")
+        import vllm.envs
+
+        vllm.envs.__dict__.pop("VLLM_P2P_IDLE_TIMEOUT_S", None)
+
+        mgr = self._make_with_fake_data()
+        old = _FakeSession(
+            peer_id="old:1",
+            last_activity_at=time.monotonic() - 999999,
+        )
+        mgr._sessions["old:1"] = old  # type: ignore[assignment]
+
+        mgr._reap_idle_sessions()
+        assert "old:1" in mgr._sessions

tests/v1/kv_offload/tiering/p2p/test_sessions.py

@@ -2698,3 +2698,141 @@ class TestTransferDoneMsgValidation:
         }
         with pytest.raises(ValueError, match="success"):
             TransferDoneMsg.validate(msg)
+
+
+# ---------------------------------------------------------------------------
+# Tests for handshake deadline
+# ---------------------------------------------------------------------------
+
+
+class TestHandshakeDeadline:
+    """A connected-but-not-ready session is marked dead after the
+    handshake timeout expires."""
+
+    def test_marks_dead_after_timeout(self, monkeypatch):
+        monkeypatch.setenv("VLLM_P2P_HANDSHAKE_TIMEOUT_S", "1")
+        import vllm.envs
+
+        vllm.envs.__dict__.pop("VLLM_P2P_HANDSHAKE_TIMEOUT_S", None)
+
+        transport = FakeDataTransport()
+        conn = FakeConnection(peer_id="stale:9999")
+        session = P2PSession(
+            peer_id="stale:9999",
+            local_id="local:1",
+            transport=transport,
+            local_block_len=4096,
+            local_hash_seed=_DEFAULT_HASH_SEED,
+            conn=conn,
+        )
+
+        assert session.connected
+        assert not session.ready
+        assert session.alive
+
+        session._connected_at = time.monotonic() - 5
+
+        session.poll()
+        assert not session.alive
+
+    def test_no_timeout_when_ready(self, monkeypatch):
+        monkeypatch.setenv("VLLM_P2P_HANDSHAKE_TIMEOUT_S", "1")
+        import vllm.envs
+
+        vllm.envs.__dict__.pop("VLLM_P2P_HANDSHAKE_TIMEOUT_S", None)
+
+        transport = FakeDataTransport()
+        conn = FakeConnection(peer_id="good:1234")
+        session = P2PSession(
+            peer_id="good:1234",
+            local_id="local:1",
+            transport=transport,
+            local_block_len=4096,
+            local_hash_seed=_DEFAULT_HASH_SEED,
+            conn=conn,
+        )
+
+        # Complete the handshake by feeding ConnectMsg + ConnectAckMsg.
+        conn._inbox = [
+            {
+                TYPE_KEY: ConnectMsg.TYPE,
+                ConnectMsg.PEER_ID: "good:1234",
+                ConnectMsg.AGENT_METADATA: b"meta",
+                ConnectMsg.BASE_ADDR: 0x1000,
+                ConnectMsg.NUM_BLOCKS: 16,
+                ConnectMsg.BLOCK_LEN: 4096,
+                ConnectMsg.CONFIG_FINGERPRINT: "",
+                ConnectMsg.HASH_SEED: _DEFAULT_HASH_SEED,
+            },
+            {TYPE_KEY: ConnectAckMsg.TYPE, ConnectAckMsg.PEER_ID: "good:1234"},
+        ]
+        session.poll()
+        assert session.ready
+
+        session._connected_at = time.monotonic() - 100
+        session.poll()
+        assert session.alive
+
+    def test_buffered_ack_at_deadline_completes_handshake(self, monkeypatch):
+        """A ConnectAck already in the inbox at the deadline must complete
+        the handshake instead of being discarded by mark_dead."""
+        monkeypatch.setenv("VLLM_P2P_HANDSHAKE_TIMEOUT_S", "1")
+        import vllm.envs
+
+        vllm.envs.__dict__.pop("VLLM_P2P_HANDSHAKE_TIMEOUT_S", None)
+
+        transport = FakeDataTransport()
+        conn = FakeConnection(peer_id="late-ack:1")
+        session = P2PSession(
+            peer_id="late-ack:1",
+            local_id="local:1",
+            transport=transport,
+            local_block_len=4096,
+            local_hash_seed=_DEFAULT_HASH_SEED,
+            conn=conn,
+        )
+        session._connected_at = time.monotonic() - 5
+        conn._inbox = [
+            {
+                TYPE_KEY: ConnectMsg.TYPE,
+                ConnectMsg.PEER_ID: "late-ack:1",
+                ConnectMsg.AGENT_METADATA: b"meta",
+                ConnectMsg.BASE_ADDR: 0x1000,
+                ConnectMsg.NUM_BLOCKS: 16,
+                ConnectMsg.BLOCK_LEN: 4096,
+                ConnectMsg.CONFIG_FINGERPRINT: "",
+                ConnectMsg.HASH_SEED: _DEFAULT_HASH_SEED,
+            },
+            {TYPE_KEY: ConnectAckMsg.TYPE, ConnectAckMsg.PEER_ID: "late-ack:1"},
+        ]
+        session.poll()
+        assert session.ready
+        assert session.alive
+
+    def test_activity_timestamp_updated_on_recv(self):
+        transport = FakeDataTransport()
+        conn = FakeConnection(peer_id="peer:1")
+        session = P2PSession(
+            peer_id="peer:1",
+            local_id="local:1",
+            transport=transport,
+            local_block_len=4096,
+            local_hash_seed=_DEFAULT_HASH_SEED,
+            conn=conn,
+        )
+        before = session._last_activity_at
+
+        conn._inbox = [
+            {
+                TYPE_KEY: ConnectMsg.TYPE,
+                ConnectMsg.PEER_ID: "peer:1",
+                ConnectMsg.AGENT_METADATA: b"meta",
+                ConnectMsg.BASE_ADDR: 0x1000,
+                ConnectMsg.NUM_BLOCKS: 16,
+                ConnectMsg.BLOCK_LEN: 4096,
+                ConnectMsg.CONFIG_FINGERPRINT: "",
+                ConnectMsg.HASH_SEED: _DEFAULT_HASH_SEED,
+            },
+        ]
+        session.poll()
+        assert session._last_activity_at >= before

tests/v1/kv_offload/tiering/p2p/test_zmq_transport.py

@@ -336,3 +336,47 @@ class TestZmqReconnect:
         finally:
             transport_a.close()
             transport_b.close()
+
+
+class TestExpandedMonitorEvents:
+    """The monitor mask includes EVENT_DISCONNECTED and EVENT_CLOSED so
+    both are treated as fatal."""
+
+    def test_fatal_events_include_disconnected_and_closed(self):
+        from vllm.v1.kv_offload.tiering.p2p.control.zmq import (
+            _FATAL_MONITOR_EVENTS,
+        )
+
+        assert zmq.EVENT_DISCONNECTED in _FATAL_MONITOR_EVENTS
+        assert zmq.EVENT_CLOSED in _FATAL_MONITOR_EVENTS
+
+    def test_monitor_mask_covers_fatal_events(self):
+        from vllm.v1.kv_offload.tiering.p2p.control.zmq import (
+            _FATAL_MONITOR_EVENTS,
+            _MONITOR_EVENT_MASK,
+        )
+
+        for ev in _FATAL_MONITOR_EVENTS:
+            assert _MONITOR_EVENT_MASK & ev == ev
+
+    def test_disconnect_detected_after_peer_shutdown(self):
+        """When a peer's transport closes, the outbound connection is
+        eventually marked dead by the monitor."""
+        transport_a, port_a = _make_transport()
+        transport_b, _ = _make_transport()
+
+        try:
+            conn_out = transport_b.connect(f"127.0.0.1:{port_a}")
+            conn_out.send({"type": "hello"})
+            _wait_for_inbound(transport_a)
+
+            transport_a.close()
+
+            end = time.monotonic() + 15
+            while conn_out.alive and time.monotonic() < end:
+                transport_b.poll()
+                time.sleep(0.05)
+
+            assert not conn_out.alive
+        finally:
+            transport_b.close()

vllm/envs.py

@@ -233,6 +233,9 @@ if TYPE_CHECKING:
     VLLM_NIXL_SIDE_CHANNEL_PORT: int = 5600
     VLLM_P2P_SIDE_CHANNEL_HOST: str = "localhost"
     VLLM_P2P_SIDE_CHANNEL_PORT: int = 5710
+    VLLM_P2P_MAX_PEERS: int = 128
+    VLLM_P2P_HANDSHAKE_TIMEOUT_S: int = 30
+    VLLM_P2P_IDLE_TIMEOUT_S: int = 300
     VLLM_EC_SIDE_CHANNEL_HOST: str = "localhost"
     VLLM_EC_SIDE_CHANNEL_PORT: int = 5601
     VLLM_MOONCAKE_BOOTSTRAP_PORT: int = 8998
@@ -1739,6 +1742,21 @@ environment_variables: dict[str, Callable[[], Any]] = {
     "VLLM_P2P_SIDE_CHANNEL_PORT": lambda: int(
         os.getenv("VLLM_P2P_SIDE_CHANNEL_PORT", "5710")
     ),
+    # Maximum number of concurrent P2P peer sessions. Limits ZMQ socket
+    # consumption to prevent denial-of-service via unbounded peer
+    # creation from client-supplied kv_transfer_params. Default: 128.
+    "VLLM_P2P_MAX_PEERS": lambda: int(os.getenv("VLLM_P2P_MAX_PEERS", "128")),
+    # Seconds to wait for a P2P peer handshake to complete (connected
+    # but not yet ready). Sessions exceeding this deadline are reaped.
+    # Prevents never-responding peers from accumulating indefinitely.
+    "VLLM_P2P_HANDSHAKE_TIMEOUT_S": lambda: int(
+        os.getenv("VLLM_P2P_HANDSHAKE_TIMEOUT_S", "30")
+    ),
+    # Seconds a P2P session may remain idle (no pending work, no
+    # messages) before it is eligible for eviction. Idle eviction runs
+    # only when remaining peer slots are within 8 of VLLM_P2P_MAX_PEERS.
+    # Default: 300. Set to 0 to disable.
+    "VLLM_P2P_IDLE_TIMEOUT_S": lambda: int(os.getenv("VLLM_P2P_IDLE_TIMEOUT_S", "300")),
     # IP address used for the EC connector's ZMQ side channel
     # (producer ROUTER bind, consumer DEALER dial).
     "VLLM_EC_SIDE_CHANNEL_HOST": lambda: os.getenv(

vllm/v1/kv_offload/tiering/p2p/control/zmq.py

@@ -27,11 +27,35 @@ _HEARTBEAT_IVL_MS = 2000
 _HEARTBEAT_TIMEOUT_MS = 10000
 _HEARTBEAT_TTL_MS = 10000
 
+# Events that indicate a fatal connection failure — the peer will never
+# become reachable through this socket.
+_FATAL_MONITOR_EVENTS: frozenset[int] = frozenset()  # populated after import
+
 # Shared sentinels returned when there is nothing to report.
 _EMPTY_INBOX: tuple[dict, ...] = ()
 _EMPTY_NEW_CONNECTIONS: tuple[ControlConnection, ...] = ()
 
 
+def _build_fatal_events() -> frozenset[int]:
+    events = {zmq.EVENT_DISCONNECTED, zmq.EVENT_CLOSED}
+    for attr in (
+        "EVENT_HANDSHAKE_FAILED_AUTH",
+        "EVENT_HANDSHAKE_FAILED_PROTOCOL",
+        "EVENT_HANDSHAKE_FAILED_NO_DETAIL",
+    ):
+        val = getattr(zmq, attr, None)
+        if val is not None:
+            events.add(val)
+    return frozenset(events)
+
+
+_FATAL_MONITOR_EVENTS = _build_fatal_events()
+
+_MONITOR_EVENT_MASK: int = 0
+for _ev in _FATAL_MONITOR_EVENTS:
+    _MONITOR_EVENT_MASK |= _ev
+
+
 def _tcp_addr(host: str, port: int | str) -> str:
     return f"tcp://{host}:{port}"
 
@@ -247,7 +271,7 @@ class ZmqTransport(ControlTransport):
         safe_id = peer_id.replace(":", "-").replace("/", "-")
         monitor_addr = f"inproc://p2p-monitor-{safe_id}-{self._monitor_seq}"
         self._monitor_seq += 1
-        dealer.monitor(monitor_addr, zmq.EVENT_DISCONNECTED)
+        dealer.monitor(monitor_addr, _MONITOR_EVENT_MASK)
 
         monitor_sock = self._zmq_ctx.socket(zmq.PAIR)
         monitor_sock.connect(monitor_addr)
@@ -319,29 +343,33 @@ class ZmqTransport(ControlTransport):
                 self._pending_inbound.append((sender_id, msg))
 
     def _check_monitors(self) -> None:
-        """Non-blocking: check all monitor sockets for disconnection."""
+        """Non-blocking: check all monitor sockets for fatal events."""
         for conn in self._connections.values():
             if not conn.alive:
                 continue
-            try:
-                event = zmq.utils.monitor.recv_monitor_message(
-                    conn.monitor_socket, zmq.NOBLOCK
-                )
-            except zmq.Again:
-                continue
-            except zmq.ZMQError as exc:
-                logger.warning(
-                    "ZmqTransport %s: monitor error for peer %s: %s",
-                    self._local_id,
-                    conn.peer_id,
-                    exc,
-                )
-                continue
+            while True:
+                try:
+                    event = zmq.utils.monitor.recv_monitor_message(
+                        conn.monitor_socket, zmq.NOBLOCK
+                    )
+                except zmq.Again:
+                    break
+                except zmq.ZMQError as exc:
+                    logger.warning(
+                        "ZmqTransport %s: monitor error for peer %s: %s",
+                        self._local_id,
+                        conn.peer_id,
+                        exc,
+                    )
+                    break
 
-            if event["event"] == zmq.EVENT_DISCONNECTED:
-                logger.debug(
-                    "ZmqTransport %s: peer %s disconnected",
-                    self._local_id,
-                    conn.peer_id,
-                )
-                conn.mark_dead()
+                ev = event["event"]
+                if ev in _FATAL_MONITOR_EVENTS:
+                    logger.debug(
+                        "ZmqTransport %s: peer %s fatal event 0x%x",
+                        self._local_id,
+                        conn.peer_id,
+                        ev,
+                    )
+                    conn.mark_dead()
+                    break

vllm/v1/kv_offload/tiering/p2p/manager.py

@@ -69,6 +69,11 @@ _SHUTDOWN_DRAIN_TIMEOUT_S = 3.0
 # enough to avoid busy-spinning the scheduler thread.
 _DRAIN_SLEEP_S = 0.001
 
+# Idle sessions are only reaped when remaining capacity is at most this
+# many slots. Session setup is expensive; below this headroom, keep idle
+# peers so a later burst does not pay handshake again.
+_IDLE_REAP_HEADROOM = 8
+
 
 def _remote_prefiller_params(kv_params: dict | None) -> dict | None:
     """Return the ``remote_prefiller`` sub-dict, or None if absent.
@@ -428,7 +433,9 @@ class P2PSecondaryTierManager(SecondaryTierManager):
         _annotate_req_context(req_context)
         source = req_context.get_state(P2PSourceInfo)
         if source is not None:
-            self._get_or_create_session(source.peer_id)
+            session = self._get_or_create_session(source.peer_id)
+            if session is None:
+                self._failed_req_ids.add(source.kv_request_id)
         dest = req_context.get_state(P2PDestInfo)
         if dest is not None and dest.kv_request_id:
             return RequestOffloadingContext(policy=OffloadPolicy.REQUEST_LEVEL)
@@ -685,38 +692,78 @@ class P2PSecondaryTierManager(SecondaryTierManager):
             self._hash_seed = get_none_hash_seed()
         return self._hash_seed
 
-    def _get_or_create_session(self, peer_id: str) -> P2PSession:
+    def _get_or_create_session(self, peer_id: str) -> P2PSession | None:
         """Return the existing session for peer_id, or open one outbound.
 
         Consumer-side helper for on_new_request: when ``remote_prefiller``
-        (PD) or ``remote_kv_source`` (symmetric P2P) is set, the consumer must reach the
-        producer at peer_id. If we already have a session toward that
-        peer (from a prior load or a peer-initiated inbound), reuse it;
-        otherwise open an outbound ControlConnection and build a
+        (PD) or ``remote_kv_source`` (symmetric P2P) is set, the consumer
+        must reach the producer at peer_id. If we already have a session
+        toward that peer (from a prior load or a peer-initiated inbound),
+        reuse it; otherwise open an outbound ControlConnection and build a
         connected session.
+
+        Returns None if the session could not be created (capacity
+        exceeded or transport error).
         """
         session = self._sessions.get(peer_id)
         if session is not None:
             return session
-        conn = self._control.connect(peer_id)
-        session = P2PSession(
-            peer_id=peer_id,
-            local_id=self._local_id,
-            transport=self._data,
-            local_block_len=self._data.block_len,
-            local_hash_seed=self._get_hash_seed(),
-            conn=conn,
-        )
+
+        max_peers = envs.VLLM_P2P_MAX_PEERS
+        if len(self._sessions) >= max_peers:
+            logger.warning(
+                "P2P %s: rejecting outbound peer %s — session limit "
+                "reached (%d/%d). Set VLLM_P2P_MAX_PEERS to increase.",
+                self._local_id,
+                peer_id,
+                len(self._sessions),
+                max_peers,
+            )
+            return None
+
+        conn = None
+        try:
+            conn = self._control.connect(peer_id)
+            session = P2PSession(
+                peer_id=peer_id,
+                local_id=self._local_id,
+                transport=self._data,
+                local_block_len=self._data.block_len,
+                local_hash_seed=self._get_hash_seed(),
+                conn=conn,
+            )
+        except Exception:
+            logger.exception(
+                "P2P %s: failed to open connection to %s",
+                self._local_id,
+                peer_id,
+            )
+            if conn is not None:
+                conn.close()
+            return None
+
         self._sessions[peer_id] = session
         return session
 
     def _accept_new_peers(self, new_connections: Sequence[ControlConnection]) -> None:
+        max_peers = envs.VLLM_P2P_MAX_PEERS
         for conn in new_connections:
             logger.info(
                 "P2P %s: accepting incoming connection from %s",
                 self._local_id,
                 conn.peer_id,
             )
+            if len(self._sessions) >= max_peers:
+                logger.warning(
+                    "P2P %s: rejecting inbound peer %s — session limit "
+                    "reached (%d/%d). Set VLLM_P2P_MAX_PEERS to increase.",
+                    self._local_id,
+                    conn.peer_id,
+                    len(self._sessions),
+                    max_peers,
+                )
+                conn.close()
+                continue
             try:
                 existing = self._sessions.get(conn.peer_id)
                 if existing is not None:
@@ -827,6 +874,53 @@ class P2PSecondaryTierManager(SecondaryTierManager):
                 len(batches),
             )
 
+    def _reap_idle_sessions(self) -> None:
+        """Evict idle sessions when remaining peer slots are scarce.
+
+        A session is idle when it has no pending work and its last activity
+        timestamp exceeds ``VLLM_P2P_IDLE_TIMEOUT_S``. Eviction runs only
+        when ``len(_sessions) >= VLLM_P2P_MAX_PEERS - _IDLE_REAP_HEADROOM``
+        so a quiet but healthy mesh is not torn down. Uses the same
+        teardown path as ``_reap_dead_sessions``.
+        """
+        idle_timeout = envs.VLLM_P2P_IDLE_TIMEOUT_S
+        if idle_timeout <= 0:
+            return
+        max_peers = envs.VLLM_P2P_MAX_PEERS
+        headroom = min(_IDLE_REAP_HEADROOM, max(0, max_peers - 1))
+        if len(self._sessions) < max_peers - headroom:
+            return
+        deadline = time.monotonic() - idle_timeout
+        idle: list[str] | None = None
+        for pid, s in self._sessions.items():
+            if not s.has_pending_work and s.last_activity_at <= deadline:
+                if idle is None:
+                    idle = []
+                idle.append(pid)
+        if idle is None:
+            return
+        for pid in idle:
+            session = self._sessions.pop(pid)
+            stale_kv_ids = [
+                kid for kid, s in self._kv_to_session.items() if s is session
+            ]
+            for kid in stale_kv_ids:
+                del self._kv_to_session[kid]
+            close_result = session.close()
+            for job_id in close_result.failed_jobs:
+                self._finished_jobs.append(JobResult(job_id=job_id, success=False))
+            for job_id in close_result.failed_stores:
+                self._finished_jobs.append(JobResult(job_id=job_id, success=False))
+            self._failed_req_ids.update(close_result.failed_req_ids)
+            self._failed_serve_ctxs.extend(close_result.failed_serves)
+            self._data.remove_remote_peer(pid)
+            logger.info(
+                "P2P %s: evicted idle peer %s (idle > %ds)",
+                self._local_id,
+                pid,
+                idle_timeout,
+            )
+
     # ------------------------------------------------------------------
     # Polling
     # ------------------------------------------------------------------
@@ -893,6 +987,7 @@ class P2PSecondaryTierManager(SecondaryTierManager):
 
         self._reap_dead_sessions()
         self._reap_unbound_stores()
+        self._reap_idle_sessions()
 
     # ------------------------------------------------------------------
     # Lifecycle

vllm/v1/kv_offload/tiering/p2p/session/session.py

@@ -18,9 +18,11 @@ received our ConnectMsg, after which queued outgoing messages are flushed.
 from __future__ import annotations
 
 import contextlib
+import time
 from collections.abc import Sequence
 from typing import TYPE_CHECKING, NamedTuple
 
+import vllm.envs as envs
 from vllm.logger import init_logger
 from vllm.v1.kv_offload.base import OffloadKey
 from vllm.v1.kv_offload.tiering.p2p.control.base import ControlConnection
@@ -128,6 +130,10 @@ class P2PSession:
         # Consecutive non-protocol dispatch errors. Reset on success.
         self._dispatch_error_count: int = 0
 
+        now = time.monotonic()
+        self._connected_at: float | None = None
+        self._last_activity_at: float = now
+
         # kv_request_ids whose FetchMsg arrived during the current poll
         # tick. Drained and returned in the next poll() result so the
         # manager can bind kv_request_id → session and replay any
@@ -168,6 +174,11 @@ class P2PSession:
         """True while inbound loads or outbound transfers are outstanding."""
         return self._client.has_active_loads or self._server.has_inflight_transfers
 
+    @property
+    def last_activity_at(self) -> float:
+        """Monotonic timestamp of the last meaningful session activity."""
+        return self._last_activity_at
+
     # ------------------------------------------------------------------
     # Connection lifecycle
     # ------------------------------------------------------------------
@@ -181,6 +192,9 @@ class P2PSession:
         if self._conn is not None:
             raise ValueError(f"P2PSession {self.peer_id}: already connected")
         self._conn = conn
+        now = time.monotonic()
+        self._connected_at = now
+        self._last_activity_at = now
         self._send_connect()
 
     # ------------------------------------------------------------------
@@ -249,17 +263,39 @@ class P2PSession:
     def poll(self) -> SessionPollResult:
         """Process incoming messages, drive transfers, apply timeouts."""
         if self._conn is None:
-            # Pending session — store-job timeouts still apply so buffered
-            # jobs that never get picked up are surfaced as failures.
             return SessionPollResult(
                 loads=[],
                 stores=self._server.collect_idle_timeouts(),
                 new_fetch_ids=[],
             )
 
-        for msg in self._conn.recv():
+        # Drain the inbox first so a ConnectAck already buffered at the
+        # deadline (e.g. after a scheduler stall) can complete the
+        # handshake instead of being discarded by mark_dead.
+        messages = self._conn.recv()
+        if messages:
+            self._last_activity_at = time.monotonic()
+        for msg in messages:
             self._on_message(msg)
 
+        # Handshake deadline: a peer that never completes the handshake
+        # is stuck in connected-but-not-ready state. Mark it dead so the
+        # manager reaps it instead of accumulating sockets indefinitely.
+        if (
+            not self._send_ready
+            and self._conn.alive
+            and self._connected_at is not None
+            and time.monotonic() - self._connected_at
+            > envs.VLLM_P2P_HANDSHAKE_TIMEOUT_S
+        ):
+            logger.warning(
+                "P2PSession %s: handshake timed out after %ds "
+                "— marking connection dead",
+                self.peer_id,
+                envs.VLLM_P2P_HANDSHAKE_TIMEOUT_S,
+            )
+            self._conn.mark_dead()
+
         loads = self._client.collect_results()
         stores = self._server.collect_results()
         self._server.drain_pending_aborts()
@@ -533,6 +569,7 @@ class P2PSession:
             return
         try:
             self._conn.send(msg)
+            self._last_activity_at = time.monotonic()
             logger.debug("P2PSession %s: sent %s", self.peer_id, msg.get(TYPE_KEY))
         except Exception:
             # A send failure means the connection is broken. Swallowing it