perf(http-pd): use shared SseEncoder for logprob-merge SSE re-encode

Replace the per-chunk `serde_json::to_string` + `format!("data: …")`
pair in `merge_streaming_logprobs` with the shared
`common::sse::SseEncoder`. A single encoder is created once per stream
in the `create_streaming_response` relay task and threaded into the
merge helper, so the serialization buffer is reused across every
re-encoded chunk (1 allocation per chunk instead of 2).

The idiosyncratic streaming-error frame in `handle_decode_error_response`
(`data: {'error': …}` without a trailing blank line) is left untouched to
avoid changing its wire format. No behavior change.

Signed-off-by: XinyueZhang369 <zoeyzhang369@gmail.com>

Author: XinyueZhang369

model_gateway/src/routers/http/pd_router.rs

@@ -36,6 +36,7 @@ use crate::{
         common::{
             header_utils,
             retry::{is_retryable_status, RetryExecutor},
+            sse::SseEncoder,
         },
         error,
         grpc::utils::{error_type_from_status, route_to_endpoint},
@@ -910,14 +911,20 @@ impl PDRouter {
         )]
         tokio::spawn(async move {
             futures_util::pin_mut!(stream);
+            // Reusable SSE encoder for the logprob-merge re-encode path.
+            let mut encoder = SseEncoder::new();
             while let Some(chunk_result) = stream.next().await {
                 match chunk_result {
                     Ok(chunk) => {
                         let is_done = memmem::find(&chunk, b"data: [DONE]").is_some();
 
                         let result = if return_logprob && prefill_logprobs.is_some() {
-                            Self::merge_streaming_logprobs(prefill_logprobs.clone(), &chunk)
-                                .unwrap_or(chunk)
+                            Self::merge_streaming_logprobs(
+                                prefill_logprobs.clone(),
+                                &chunk,
+                                &mut encoder,
+                            )
+                            .unwrap_or(chunk)
                         } else {
                             chunk
                         };
@@ -1138,6 +1145,7 @@ impl PDRouter {
     fn merge_streaming_logprobs(
         prefill_logprobs: Option<Value>,
         decode_chunk: &[u8],
+        encoder: &mut SseEncoder,
     ) -> Result<bytes::Bytes, ()> {
         // Skip non-data chunks
         let chunk_str = std::str::from_utf8(decode_chunk).map_err(|_| ())?;
@@ -1168,12 +1176,8 @@ impl PDRouter {
             }
         }
 
-        // Re-serialize
-        let merged_str = format!(
-            "data: {}\n\n",
-            serde_json::to_string(&decode_json).unwrap_or_default()
-        );
-        Ok(bytes::Bytes::from(merged_str))
+        // Re-serialize via the shared encoder (reuses its buffer across chunks).
+        encoder.encode_data(&decode_json).map_err(|_| ())
     }
 }