fix(disagg): dedup per-layer enqueues so high-cache-hit can't hang (lever A)

E2E at high cache-hit + concurrency exposed a vicious cycle: a context stays active
until finish, but the notifier re-fires every layer on every subsequent forward and
note_enqueued counted each, so _enqueued (the finish target) grew by the layer count
each forward (target=3042-5538 observed) faster than 4 workers can drain -> finish
never reaches it -> 30s timeout -> context stays active -> repeat. TTFT p99 = 116s.
Fix: note_enqueued(layer_id) dedups per layer (target caps at the layer count); the
first fire for a layer is from the request's own forward so its event is correct.
Also guard register() against overwriting an active room (was leaking + re-registering,
registered=917 for ~100 reqs).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

python/sglang/srt/disaggregation/cp_per_layer_transfer.py

@@ -81,6 +81,7 @@ class PerLayerTransferContext:
         self.num_layers = num_layers
         self._batch_ids: List[int] = []
         self._submitted_layers: set[int] = set()
+        self._enqueued_layers: set[int] = set()
         self._enqueued = 0
         self._processed = 0
         self._failed = False
@@ -91,11 +92,19 @@ class PerLayerTransferContext:
         with self._cond:
             return self._failed
 
-    def note_enqueued(self) -> None:
-        """Called by the manager when it enqueues a (ctx, layer) item — the true count
-        of notifier fires for this request, which finish() waits on."""
+    def note_enqueued(self, layer_id: int) -> bool:
+        """Reserve layer_id for enqueue. Returns True only the FIRST time this layer is
+        seen, so the manager enqueues + counts each layer ONCE. Critical: the notifier
+        re-fires every layer on every subsequent forward while this ctx is still active
+        (awaiting finish); without dedup, _enqueued would grow unbounded (target>>layers)
+        and finish() could never reach it -> 30s timeouts. The first fire for a layer is
+        from THIS request's own forward, so its event is the correct one."""
         with self._cond:
+            if layer_id in self._enqueued_layers:
+                return False
+            self._enqueued_layers.add(layer_id)
             self._enqueued += 1
+            return True
 
     def _submit_one(self, layer_id: int) -> None:
         """Submit layer_id's transfer (KV assumed written). Updates state; logs +
@@ -222,6 +231,8 @@ class PerLayerTransferManager:
 
     def register(self, room, ctx: PerLayerTransferContext) -> None:
         with self._active_lock:
+            if room in self._active:
+                return  # already registered for its forward; don't overwrite (leak)
             self._active[room] = ctx
 
     def has_active(self) -> bool:
@@ -242,8 +253,8 @@ class PerLayerTransferManager:
             else:
                 event.record()
         for ctx in ctxs:
-            ctx.note_enqueued()
-            self._q.put((ctx, layer_id, event))
+            if ctx.note_enqueued(layer_id):  # enqueue each layer ONCE per ctx
+                self._q.put((ctx, layer_id, event))
 
     def has_room(self, room) -> bool:
         with self._active_lock: