@tduncan / Nvidia-Optimization-Stack-VLLM / commits / ea84051819

[KVConnector][NIXL] Count completion notifications that arrive after KV expiry (#58875)

Signed-off-by: Zijing Liu <liuzijing2014@gmail.com>

Zijing Liu committed Oct 1, 2026 at 14:30 UTC ea84051819c2a52a535e4b6df1fd828def5880f7

7 files changed +69 -1

docs/features/nixl_connector_usage.md

@@ -454,6 +454,7 @@ exported when NixlConnector is active:

454 | `vllm:nixl_num_failed_transfers` | Counter | Cumulative count of failed NIXL KV-block transfers, including handshake and notification (`send_notif`) failures. These are grouped because all are sporadic lower-transport-layer events. |

455 | `vllm:nixl_num_failed_notifications` | Counter | Cumulative count of failed notifications (`send_notif`), retained for compatibility. These failures are also included in `vllm:nixl_num_failed_transfers`; do not sum the two counters. |

456 | `vllm:nixl_num_kv_expired_reqs` | Counter | Requests whose KV blocks expired on the prefiller before the decoder read them (tracked on the P instance). Kept separate from the failure counters above: KV expiry is an autoscaler/lease-tuning signal, not a transport issue. |

457 +| `vllm:nixl_num_notifications_after_expiry` | Counter | Completion notifications that arrive for a request that is no longer tracked, usually because its KV lease expired first. Pull connector: counted on P, where it means the decoder may have read blocks that were already reused; each matches a "Potentially invalid KV blocks" error log. Push connector: counted on whichever instance tracks the request as neither a send nor a receive (P after the lease expired, or D after the receive completed or was abandoned); each matches an "Unrecognized request" error log. Counted per notification, not per request: each remote rank notifies separately, so with TP > 1 or heterogeneous TP one late request counts several times, and the value is not comparable to `vllm:nixl_num_kv_expired_reqs`. |

458

459 !!! tip

460 High `vllm:nixl_num_kv_expired_reqs` indicates that the prefiller's lease

tests/v1/kv_connector/unit/test_multi_connector.py

+12

@@ -601,6 +601,7 @@ class TestMultiConnectorStats:

601 "num_failed_notifications": [],

602 "num_failed_handshakes": [],

603 "num_kv_expired_reqs": [],

604 + "num_notifications_after_expiry": [],

605 }

606 }

607

@@ -626,6 +627,7 @@ class TestMultiConnectorStats:

627 "num_failed_notifications": [],

628 "num_failed_handshakes": [],

629 "num_kv_expired_reqs": [],

630 + "num_notifications_after_expiry": [],

631 },

632 "MockConnector": {"mock_field": [1, 2, 3]},

633 }

@@ -657,6 +659,7 @@ class TestMultiConnectorStats:

659 "num_failed_notifications": [],

660 "num_failed_handshakes": [],

661 "num_kv_expired_reqs": [],

662 + "num_notifications_after_expiry": [],

663 },

664 }

665

@@ -678,6 +681,7 @@ class TestMultiConnectorStats:

681 "num_failed_notifications": [],

682 "num_failed_handshakes": [],

683 "num_kv_expired_reqs": [],

684 + "num_notifications_after_expiry": [],

685 }

686 )

687 mock_stats = MockConnectorStats(data={"mock_field": [1, 2, 3]})

@@ -709,6 +713,7 @@ class TestMultiConnectorStats:

713 "num_failed_notifications": [],

714 "num_failed_handshakes": [],

715 "num_kv_expired_reqs": [],

716 + "num_notifications_after_expiry": [],

717 }

718 )

719

@@ -742,6 +747,7 @@ class TestMultiConnectorStats:

747 "num_failed_notifications": [],

748 "num_failed_handshakes": [],

749 "num_kv_expired_reqs": [],

750 + "num_notifications_after_expiry": [],

751 },

752 "ExampleConnector": {"some_field": [1, 2, 3]},

753 }

@@ -781,6 +787,7 @@ class TestMultiConnectorStats:

787 "num_failed_notifications": [],

788 "num_failed_handshakes": [],

789 "num_kv_expired_reqs": [],

790 + "num_notifications_after_expiry": [],

791 }

792 )

793 }

@@ -798,6 +805,7 @@ class TestMultiConnectorStats:

805 "num_failed_notifications": [],

806 "num_failed_handshakes": [],

807 "num_kv_expired_reqs": [],

808 + "num_notifications_after_expiry": [],

809 }

810 )

811 }

@@ -831,6 +839,7 @@ class TestMultiConnectorStats:

839 "num_failed_notifications": [],

840 "num_failed_handshakes": [],

841 "num_kv_expired_reqs": [],

842 + "num_notifications_after_expiry": [],

843 }

844 )

845 }

@@ -859,6 +868,7 @@ class TestMultiConnectorStats:

868 "num_failed_notifications": [],

869 "num_failed_handshakes": [],

870 "num_kv_expired_reqs": [],

871 + "num_notifications_after_expiry": [],

872 }

873 )

874 }

@@ -887,6 +897,7 @@ class TestMultiConnectorStats:

897 "num_failed_notifications": [],

898 "num_failed_handshakes": [],

899 "num_kv_expired_reqs": [],

900 + "num_notifications_after_expiry": [],

901 }

902 )

903 }

@@ -918,6 +929,7 @@ class TestMultiConnectorStats:

929 "num_failed_notifications": [],

930 "num_failed_handshakes": [],

931 "num_kv_expired_reqs": [],

932 + "num_notifications_after_expiry": [],

933 }

934 )

935 }

tests/v1/kv_connector/unit/test_nixl_connector.py

+30

@@ -1698,6 +1698,7 @@ def test_kv_connector_stats_failure_grouping():

1698 stats.record_failed_handshake()

1699 stats.record_failed_notification()

1700 stats.record_kv_expired_req()

1701 + stats.record_notification_after_expiry()

1702 assert not stats.is_empty()

1703

1704 # No successful transfers: latency stats are zero but the failure

@@ -1706,6 +1707,7 @@ def test_kv_connector_stats_failure_grouping():

1707 assert reduced["Num successful transfers"] == 0

1708 assert reduced["Num failed transfers"] == 3

1709 assert reduced["Num KV expired reqs"] == 1

1710 + assert reduced["Num notifs after expiry"] == 1

1711

1712

1713 def test_nixl_prom_metrics_group_handshake_with_transfer_failures():

@@ -1750,6 +1752,7 @@ def test_nixl_prom_metrics_group_handshake_with_transfer_failures():

1752 stats.record_failed_handshake()

1753 stats.record_failed_notification()

1754 stats.record_kv_expired_req()

1755 + stats.record_notification_after_expiry()

1756 prom.observe(stats.data, engine_idx=0)

1757

1758 def counter_value(name: str) -> float:

@@ -1761,6 +1764,33 @@ def test_nixl_prom_metrics_group_handshake_with_transfer_failures():

1764

1765 assert counter_value("vllm:nixl_num_failed_transfers_total") == 3.0

1766 assert counter_value("vllm:nixl_num_kv_expired_reqs_total") == 1.0

1767 + assert counter_value("vllm:nixl_num_notifications_after_expiry_total") == 1.0

1768 +

1769 +

1770 +@patch(

1771 + "vllm.distributed.kv_transfer.kv_connector.v1.nixl.base_worker.NixlWrapper",

1772 + FakeNixlWrapper,

1773 +)

1774 +def test_notification_after_expiry_is_counted(default_vllm_config, dist_init):

1775 + vllm_config = create_vllm_config()

1776 + connector = NixlConnector(

1777 + vllm_config, KVConnectorRole.WORKER, make_kv_cache_config(block_size=16)

1778 + )

1779 + connector.connector_worker = FakeNixlConnectorWorker(

1780 + vllm_config, connector.engine_id, hand_shake_latency=0

1781 + )

1782 + worker = connector.connector_worker

1783 + worker._reqs_to_process.add("known")

1784 + worker._reqs_to_send["known"] = time.perf_counter() + 10

1785 + worker.nixl_wrapper.get_new_notifs = MagicMock(

1786 + return_value={"decode-agent": [b"known:1", b"unknown:1"]}

1787 + )

1788 +

1789 + assert worker._get_new_notifs() == {"known"}

1790 +

1791 + stats = connector.get_kv_connector_stats()

1792 + assert isinstance(stats, NixlKVConnectorStats)

1793 + assert stats.data["num_notifications_after_expiry"] == [1]

1794

1795

1796 def test_multi_kv_connector_stats_aggregation():

tests/v1/kv_connector/unit/test_nixl_push_connector.py

@@ -1357,6 +1357,7 @@ class TestPushWriterNegative:

1357 assert notified == set()

1358 # Did not register anywhere.

1359 assert "never-heard-of-you" not in w._recving_transfers

1360 + assert w.xfer_stats.data["num_notifications_after_expiry"] == [1]

1361

1362 def test_start_load_kv_with_empty_metadata_is_noop(self):

1363 """Empty metadata must not wake the writer or enqueue anything."""

vllm/distributed/kv_transfer/kv_connector/v1/nixl/pull_worker.py

@@ -660,6 +660,7 @@ class NixlPullConnectorWorker(NixlBaseConnectorWorker):

660 req_id not in self._reqs_to_send

661 and req_id not in self._reqs_to_process

662 ):

663 + self.xfer_stats.record_notification_after_expiry()

664 logger.error(

665 "Potentially invalid KV blocks for "

666 "unrecognized request %s were retrieved by "

vllm/distributed/kv_transfer/kv_connector/v1/nixl/push_worker.py

@@ -786,6 +786,7 @@ class NixlPushConnectorWorker(NixlBaseConnectorWorker):

786 else:

787 # Not tracked on either side (lease may have expired

788 # before the notif arrived). Log and skip.

789 + self.xfer_stats.record_notification_after_expiry()

790 logger.error(

791 "Unrecognized request %s notif (may have expired).",

792 req_id,

vllm/distributed/kv_transfer/kv_connector/v1/nixl/stats.py

+23 -1

@@ -41,6 +41,7 @@ class NixlKVConnectorStats(KVConnectorStats):

41 "num_failed_notifications": [],

42 "num_failed_handshakes": [],

43 "num_kv_expired_reqs": [],

44 + "num_notifications_after_expiry": [],

45 }

47 def record_transfer(self, res: "nixlXferTelemetry"):

@@ -66,6 +67,10 @@ class NixlKVConnectorStats(KVConnectorStats):

67 """Record a request that had its KV blocks expire."""

68 self.data["num_kv_expired_reqs"].append(1)

70 + def record_notification_after_expiry(self):

71 + """Record a completion notification that arrived after its request expired."""

72 + self.data["num_notifications_after_expiry"].append(1)

73 +

74 def clone_and_reset(self) -> "NixlKVConnectorStats":

75 old = copy.copy(self)

76 self.reset()

@@ -79,6 +84,7 @@ class NixlKVConnectorStats(KVConnectorStats):

84 and len(self.data["num_failed_notifications"]) == 0

85 and len(self.data["num_failed_handshakes"]) == 0

86 and len(self.data["num_kv_expired_reqs"]) == 0

87 + and len(self.data["num_notifications_after_expiry"]) == 0

88 )

90 def aggregate(self, other: KVConnectorStats) -> KVConnectorStats:

@@ -95,12 +101,14 @@ class NixlKVConnectorStats(KVConnectorStats):

101 # and notification failures are grouped as sporadic

102 # lower-transport-layer events, while KV expiry is reported separately

103 # as it is an actionable autoscaler signal rather than a transport

98 - # issue.

104 + # issue. Notifications arriving after expiry are surfaced too, as the

105 + # reader may have consumed KV blocks that were already reused.

106 failure_counts = {

107 "Num failed transfers": len(self.data["num_failed_transfers"])

108 + len(self.data["num_failed_handshakes"])

109 + len(self.data["num_failed_notifications"]),

110 "Num KV expired reqs": len(self.data["num_kv_expired_reqs"]),

111 + "Num notifs after expiry": len(self.data["num_notifications_after_expiry"]),

112 }

113 if self.num_successful_transfers == 0:

114 # Timing / throughput stats only cover successful transfers. If

@@ -262,6 +270,18 @@ class NixlPromMetrics(KVConnectorPromMetrics):

270 counter_nixl_num_kv_expired_reqs, self.per_engine_labelvalues

271 )

272

273 + counter_nixl_num_notifications_after_expiry = self._counter_cls(

274 + name="vllm:nixl_num_notifications_after_expiry",

275 + documentation="Number of completion notifications for requests that "

276 + "were no longer tracked, usually because their KV lease expired. The "

277 + "KV blocks may have been reused before the transfer finished. Counted "

278 + "per notification (one per remote rank), not per request.",

279 + labelnames=labelnames,

280 + )

281 + self.counter_nixl_num_notifications_after_expiry = create_metric_per_engine(

282 + counter_nixl_num_notifications_after_expiry, self.per_engine_labelvalues

283 + )

284 +

285 def observe(self, transfer_stats_data: dict[str, Any], engine_idx: int = 0):

286 for prom_obj, list_item_key in zip(

287 [

@@ -284,6 +304,7 @@ class NixlPromMetrics(KVConnectorPromMetrics):

304 self.counter_nixl_num_failed_transfers,

305 self.counter_nixl_num_failed_notifications,

306 self.counter_nixl_num_kv_expired_reqs,

307 + self.counter_nixl_num_notifications_after_expiry,

308 ],

309 [

310 # Transfer, handshake and notification failures are grouped:

@@ -297,6 +318,7 @@ class NixlPromMetrics(KVConnectorPromMetrics):

318 ),

319 ("num_failed_notifications",),

320 ("num_kv_expired_reqs",),

321 + ("num_notifications_after_expiry",),

322 ],

323 ):

324 for counter_item_key in counter_item_keys: