diff --git a/python/sglang/srt/disaggregation/encode_receiver.py b/python/sglang/srt/disaggregation/encode_receiver.py index 6047b434b..1a5028ef9 100644 --- a/python/sglang/srt/disaggregation/encode_receiver.py +++ b/python/sglang/srt/disaggregation/encode_receiver.py @@ -343,7 +343,12 @@ class MultiModalEmbeddingData(EmbeddingData): return kwargs def add(self, embedding_data: EmbeddingData): - assert self.req_id == embedding_data.req_id + if self.req_id != embedding_data.req_id: + logger.warning( + f"Dropping embedding data with mismatched req_id: " + f"expected {self.req_id}, got {embedding_data.req_id}" + ) + return False assert not self.ready_list[embedding_data.part_idx] pid = embedding_data.part_idx self.ready_list[pid] = True @@ -521,6 +526,17 @@ class WaitingImageRequest: self.recv_socket.close() return + # Extract original req_id from part_req_id and drop stale payloads + # that may arrive on a reused ZMQ port after a prior request aborted. + original_req_id = extract_original_req_id(recv_obj.req_id) + if original_req_id != self.recv_req.rid: + logger.warning( + f"Dropping stale embedding data: expected rid={self.recv_req.rid}, " + f"got rid={recv_obj.req_id} (likely from ZMQ port reuse)" + ) + continue + recv_obj.req_id = original_req_id + buffer = parts[1].buffer if hasattr(parts[1], "buffer") else parts[1] recv_obj.embedding = ( torch.frombuffer(buffer, dtype=recv_obj.dtype) @@ -528,12 +544,6 @@ class WaitingImageRequest: .clone() ) - # Extract original req_id from part_req_id - part_req_id = recv_obj.req_id - original_req_id = extract_original_req_id(part_req_id) - # Update recv_obj.req_id to original for aggregation - recv_obj.req_id = original_req_id - if self.recv_embedding_data is None: self.recv_embedding_data = MultiModalEmbeddingData.from_embedding_data( recv_obj