Fix: Reject requests with a duplicate request ID which can cause server crash/hang (#19035)

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: hnyls2002 <lsyincs@gmail.com>

python/sglang/srt/managers/tokenizer_manager.py

@@ -482,6 +482,7 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi
 
         # Normalize the request
         obj.normalize_batch_and_arguments()
+        self._validate_rid(obj)
 
         self._req_stats_init(obj, request)
         if self.server_args.language_only:
@@ -737,6 +738,21 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi
             obj, input_text, input_ids, input_embeds, mm_inputs, token_type_ids
         )
 
+    def _validate_rid(self, obj: Union[GenerateReqInput, EmbeddingReqInput]) -> None:
+        """Validate the request ID (rid) uniqueness."""
+        rid = obj.rid
+        if rid is None:
+            return
+        ids = rid if isinstance(rid, list) else [rid]
+        if len(ids) != len(set(ids)):
+            raise ValueError(
+                f"Duplicate request IDs detected within the request: {ids}"
+            )
+
+        for i in ids:
+            if i in self.rid_to_state:
+                raise ValueError(f"Duplicate request ID detected: {i}")
+
     def _validate_one_request(
         self, obj: Union[GenerateReqInput, EmbeddingReqInput], input_ids: List[int]
     ) -> None:

test/registered/scheduler/test_abort.py

@@ -1,4 +1,5 @@
 import multiprocessing
+import threading
 import time
 import unittest
 from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -125,6 +126,86 @@ class TestAbortAll(AbortAllMixin, CustomTestCase):
     def tearDownClass(cls):
         kill_process_tree(cls.process.pid)
 
+    def _generate_with_rid(self, rid, max_new_tokens=8):
+        return requests.post(
+            f"{self.base_url}/generate",
+            json={
+                "text": "The capital of France is",
+                "sampling_params": {
+                    "temperature": 0,
+                    "max_new_tokens": max_new_tokens,
+                },
+                "rid": rid,
+            },
+            timeout=30,
+        )
+
+    def test_duplicate_rid_sequential_ok(self):
+        rid = "dup-rid-test-sequential"
+        resp1 = self._generate_with_rid(rid)
+        self.assertEqual(resp1.status_code, 200)
+        self.assertNotIn("error", resp1.json())
+
+        resp2 = self._generate_with_rid(rid)
+        self.assertEqual(resp2.status_code, 200)
+        self.assertNotIn("error", resp2.json())
+
+    def test_duplicate_rid_concurrent_rejected(self):
+        rid = "dup-rid-test-concurrent"
+        results = {}
+
+        def send(key, max_tokens):
+            results[key] = self._generate_with_rid(rid, max_new_tokens=max_tokens)
+
+        t1 = threading.Thread(target=send, args=("first", 512))
+        t2 = threading.Thread(target=send, args=("second", 8))
+        t1.start()
+        time.sleep(0.1)
+        t2.start()
+        t1.join(timeout=30)
+        t2.join(timeout=30)
+
+        r1, r2 = results["first"], results["second"]
+        self.assertTrue(
+            r1.status_code == 400 or r2.status_code == 400,
+            "One of the concurrent duplicate-rid requests should be rejected",
+        )
+
+        rejected = r2 if r2.status_code == 400 else r1
+        self.assertIn("Duplicate request ID", rejected.json()["error"]["message"])
+
+    def test_duplicate_rid_in_batch(self):
+        rid = "dup-rid-batch"
+        response = requests.post(
+            f"{self.base_url}/generate",
+            json={
+                "text": ["Hello", "World"],
+                "sampling_params": {"temperature": 0, "max_new_tokens": 8},
+                "rid": [rid, rid],
+            },
+            timeout=30,
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertIn("Duplicate request ID", response.json()["error"]["message"])
+
+    def test_server_healthy_after_duplicate_rid(self):
+        requests.post(
+            f"{self.base_url}/generate",
+            json={
+                "text": ["Hello", "World"],
+                "sampling_params": {"temperature": 0, "max_new_tokens": 8},
+                "rid": ["dup-health", "dup-health"],
+            },
+            timeout=30,
+        )
+
+        resp = requests.get(f"{self.base_url}/health", timeout=5)
+        self.assertEqual(resp.status_code, 200)
+
+        resp = self._generate_with_rid("after-dup-health")
+        self.assertEqual(resp.status_code, 200)
+        self.assertIn("text", resp.json())
+
 
 class TestAbortAllWithRetraction(CustomTestCase):
     @classmethod