[Bugfix][KV Offload][P2P] Drop a timed-out store job's blocks from the round's parked supply

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

@@ -1695,6 +1695,36 @@ def test_store_timeout_then_late_failure_no_duplicate(self):
             f"unexpected duplicate StoreResult after timeout: {stores}"
         )
 
+    def test_fetch_after_symmetric_supply_timed_out_reads_nothing(self):
+        """A lookup pin whose store job timed out has been reported failed,
+        so the engine unpinned the primary slot. A late FetchMsg must not
+        read that slot: the round fails fast instead."""
+        parent = FakeParent(stored={b"hA": 7})
+        session, conn, transport = _make_session()
+        _activate(session, conn)
+        _send_lookup(conn, "req-1", [b"hA"])
+        session.poll()
+        _serve(session, parent)
+        (job_id,) = session._server._store_jobs
+
+        session._server._store_jobs[job_id] = time.monotonic() - 60.0
+        assert StoreResult(job_id=job_id, success=False) in session.poll().stores
+
+        conn.enqueue(
+            {
+                TYPE_KEY: FetchMsg.TYPE,
+                FetchMsg.ROUND_SEQ: 0,
+                FetchMsg.KV_REQUEST_ID: "req-1",
+                FetchMsg.KEYS: [b"hA"],
+                FetchMsg.BLOCK_INDEXES: [20],
+            }
+        )
+        session.poll()
+
+        assert transport._transfers == {}
+        dones = [m for m in conn._sent if m[TYPE_KEY] == TransferDoneMsg.TYPE]
+        assert dones and dones[-1][TransferDoneMsg.SUCCESS] is False
+
 
 # ---------------------------------------------------------------------------
 # finish_request server-role early-fail flow

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

@@ -1135,4 +1135,17 @@ def _timeout_pending_store_jobs(self) -> list[StoreResult]:
             del self._store_jobs[jid]
             results.append(StoreResult(job_id=jid, success=False))
             logger.warning("P2PSession %s: store job %d timed out", self._peer_id, jid)
+        # A failed store job's primary slots are unpinned by the engine, so its
+        # blocks must no longer be offered to a fetch. A symmetric round then
+        # fails fast in on_fetch instead of reading a slot that may be reused.
+        expired = set(timed_out)
+        for st in self._requests.values():
+            for rnd in st.outbound.values():
+                if rnd.pending_job_ids & expired:
+                    rnd.pending_job_ids -= expired
+                    rnd.available = {
+                        key: entry
+                        for key, entry in rnd.available.items()
+                        if entry[0] not in expired
+                    }
         return results