[PD] Downgrade propagated rank failure logs from error to debug (#27534)

Signed-off-by: Shangming Cai <csmthu@gmail.com>

python/sglang/srt/disaggregation/common/conn.py

@@ -49,6 +49,22 @@ from sglang.srt.utils.network import (
 logger = logging.getLogger(__name__)
 
 
+class KVTransferError(Exception):
+    def __init__(
+        self,
+        bootstrap_room: int,
+        failure_reason: str,
+        is_from_another_rank: bool = False,
+    ):
+        super().__init__(failure_reason)
+        self.bootstrap_room = bootstrap_room
+        self.failure_reason = failure_reason
+        self.is_from_another_rank = is_from_another_rank
+
+    def __str__(self):
+        return f"KVTransferError(bootstrap_room={self.bootstrap_room}): {self.failure_reason}"
+
+
 @dataclasses.dataclass
 class PrefillServerInfo:
     # Topology fields (fetched from bootstrap server)

python/sglang/srt/disaggregation/decode.py

@@ -643,10 +643,16 @@ class DecodePreallocQueue(DecodeHiCachePreallocMixin):
                 decode_req.req.time_stats.set_bootstrap_done_time()
             elif poll == KVPoll.Failed:
                 error_message = f"Decode handshake failed for request rank={self.tp_rank} {decode_req.req.rid=} {decode_req.req.bootstrap_room=}"
+                is_propagated = False
                 try:
                     decode_req.kv_receiver.failure_exception()
                 except Exception as e:
                     error_message += f" with exception {e}"
+                    is_propagated = getattr(e, "is_from_another_rank", False)
+                # Mute error message for propagated exceptions to avoid duplicate logging
+                if is_propagated:
+                    logger.debug(error_message)
+                else:
                     logger.error(error_message)
                 prepare_abort(
                     decode_req.req,
@@ -1609,12 +1615,18 @@ class DecodeTransferQueue(DecodeHiCacheTransferMixin):
                     f"Decode transfer failed for request rank={self.tp_rank} "
                     f"{decode_req.req.rid=} {decode_req.req.bootstrap_room=}"
                 )
+                is_propagated = False
                 if poll == KVPoll.Failed:
                     try:
                         decode_req.kv_receiver.failure_exception()
                     except Exception as e:
                         error_message += f" with exception {e}"
+                        is_propagated = getattr(e, "is_from_another_rank", False)
                 self._clean_hicache_prefetch_resources(decode_req)
+                # Mute error message for propagated exceptions to avoid duplicate logging
+                if is_propagated:
+                    logger.debug(error_message)
+                else:
                     logger.error(error_message)
                 prepare_abort(
                     decode_req.req,

python/sglang/srt/disaggregation/mooncake/conn.py

@@ -20,6 +20,7 @@ from sglang.srt.disaggregation.common.conn import (
     CommonKVManager,
     CommonKVReceiver,
     CommonKVSender,
+    KVTransferError,
 )
 from sglang.srt.disaggregation.common.staging_handler import (
     DecodeStagingContext,
@@ -62,16 +63,6 @@ FAILED_SESSION_RECOVERIES = Counter(
 )
 
 
-class KVTransferError(Exception):
-    def __init__(self, bootstrap_room: int, failure_reason: str):
-        super().__init__(failure_reason)
-        self.bootstrap_room = bootstrap_room
-        self.failure_reason = failure_reason
-
-    def __str__(self):
-        return f"KVTransferError(bootstrap_room={self.bootstrap_room}): {self.failure_reason}"
-
-
 # decode
 @dataclasses.dataclass
 class TransferInfo:
@@ -1747,10 +1738,13 @@ class MooncakeKVSender(CommonKVSender):
         self.clear()
 
         with self.kv_mgr.failure_lock:
-            failure_reason = self.kv_mgr.failure_records.pop(
+            failure_reason = self.kv_mgr.failure_records.pop(self.bootstrap_room, None)
-                self.bootstrap_room, "Failed due to an unknown reason from another rank"
+        is_propagated = failure_reason is None
+        if is_propagated:
+            failure_reason = "Failed due to an unknown reason from another rank"
+        raise KVTransferError(
+            self.bootstrap_room, failure_reason, is_from_another_rank=is_propagated
         )
-        raise KVTransferError(self.bootstrap_room, failure_reason)
 
     def _init_trace_ctx(self):
         if self.kv_mgr.enable_trace:
@@ -1908,10 +1902,13 @@ class MooncakeKVReceiver(CommonKVReceiver):
         self.clear()
 
         with self.kv_mgr.failure_lock:
-            failure_reason = self.kv_mgr.failure_records.pop(
+            failure_reason = self.kv_mgr.failure_records.pop(self.bootstrap_room, None)
-                self.bootstrap_room, "Failed due to an unknown reason from another rank"
+        is_propagated = failure_reason is None
+        if is_propagated:
+            failure_reason = "Failed due to an unknown reason from another rank"
+        raise KVTransferError(
+            self.bootstrap_room, failure_reason, is_from_another_rank=is_propagated
         )
-        raise KVTransferError(self.bootstrap_room, failure_reason)
 
 
 class MooncakeKVBootstrapServer(CommonKVBootstrapServer):

python/sglang/srt/disaggregation/mori/conn.py

@@ -32,6 +32,7 @@ from sglang.srt.disaggregation.common.conn import (
     CommonKVManager,
     CommonKVReceiver,
     CommonKVSender,
+    KVTransferError,
 )
 from sglang.srt.disaggregation.common.utils import (
     AuxDataCodec,
@@ -1539,10 +1540,13 @@ class MoriKVSender(CommonKVSender):
             self._finalize_failure()
         self.clear()
         with self.kv_mgr.failure_lock:
-            failure_reason = self.kv_mgr.failure_records.pop(
+            failure_reason = self.kv_mgr.failure_records.pop(self.bootstrap_room, None)
-                self.bootstrap_room, "KV transfer failed"
+        is_propagated = failure_reason is None
+        if is_propagated:
+            failure_reason = "KV transfer failed"
+        raise KVTransferError(
+            self.bootstrap_room, failure_reason, is_from_another_rank=is_propagated
         )
-        raise RuntimeError(failure_reason)
 
     def abort(self):
         self._finalize_failure("Aborted by AbortReq.")
@@ -1675,10 +1679,13 @@ class MoriKVReceiver(CommonKVReceiver):
 
         self.clear()
         with self.kv_mgr.failure_lock:
-            failure_reason = self.kv_mgr.failure_records.pop(
+            failure_reason = self.kv_mgr.failure_records.pop(self.bootstrap_room, None)
-                self.bootstrap_room, "KV transfer failed"
+        is_propagated = failure_reason is None
+        if is_propagated:
+            failure_reason = "KV transfer failed"
+        raise KVTransferError(
+            self.bootstrap_room, failure_reason, is_from_another_rank=is_propagated
         )
-        raise RuntimeError(failure_reason)
 
     def abort(self):
         if self.bootstrap_room is None:

python/sglang/srt/disaggregation/nixl/conn.py

@@ -22,6 +22,7 @@ from sglang.srt.disaggregation.common.conn import (
     CommonKVManager,
     CommonKVReceiver,
     CommonKVSender,
+    KVTransferError,
 )
 from sglang.srt.disaggregation.common.staging_handler import StagingRegisterInfo
 from sglang.srt.disaggregation.common.utils import (
@@ -1973,8 +1974,10 @@ class NixlKVSender(CommonKVSender):
         if exc is not None:
             raise exc
         if failure_reason is not None:
-            raise RuntimeError(failure_reason)
+            raise KVTransferError(self.bootstrap_room, failure_reason)
-        raise RuntimeError("NIXL KVSender Exception")
+        raise KVTransferError(
+            self.bootstrap_room, "NIXL KVSender Exception", is_from_another_rank=True
+        )
 
 
 class NixlKVReceiver(CommonKVReceiver):
@@ -2138,7 +2141,14 @@ class NixlKVReceiver(CommonKVReceiver):
                 )
 
     def failure_exception(self):
-        raise RuntimeError("NIXL KVReceiver Exception")
+        with self.kv_mgr.failure_lock:
+            failure_reason = self.kv_mgr.failure_records.pop(self.bootstrap_room, None)
+        is_propagated = failure_reason is None
+        if is_propagated:
+            failure_reason = "NIXL KVReceiver Exception"
+        raise KVTransferError(
+            self.bootstrap_room, failure_reason, is_from_another_rank=is_propagated
+        )
 
 
 class NixlKVBootstrapServer(CommonKVBootstrapServer):

python/sglang/srt/disaggregation/prefill.py

@@ -721,10 +721,16 @@ class SchedulerDisaggregationPrefillMixin:
                 req.time_stats.set_prefill_kv_transfer_finish_time()
             elif poll == KVPoll.Failed:
                 error_message = f"Prefill transfer failed for request rank={self.ps.tp_rank} {req.rid=} {req.bootstrap_room=}"
+                is_propagated = False
                 try:
                     req.disagg_kv_sender.failure_exception()
                 except Exception as e:
                     error_message += f" with exception {e}"
+                    is_propagated = getattr(e, "is_from_another_rank", False)
+                # Mute error message for propagated exceptions to avoid duplicate logging
+                if is_propagated:
+                    logger.debug(error_message)
+                else:
                     logger.warning(error_message)
                 req.time_stats.trace_ctx.abort(abort_info={"reason": error_message})
                 release_kv_cache(req, self.tree_cache)  # unlock the tree
@@ -802,10 +808,16 @@ class SchedulerDisaggregationPrefillMixin:
             f"Prefill bootstrap failed for request rank={self.ps.tp_rank} "
             f"{req.rid=} {req.bootstrap_room=}"
         )
+        is_propagated = False
         try:
             req.disagg_kv_sender.failure_exception()
         except Exception as e:
             error_message += f" with exception {e}"
+            is_propagated = getattr(e, "is_from_another_rank", False)
+        # Mute error message for propagated exceptions to avoid duplicate logging
+        if is_propagated:
+            logger.debug(error_message)
+        else:
             logger.warning(error_message)
         req.time_stats.trace_ctx.abort(abort_info={"reason": error_message})
         if req.req_pool_idx is not None or self.tree_cache.supports_mamba():