限制原始图片解码并发

为 raw image PNG 预处理增加 AppState 专用 semaphore。

按请求截止时间限制槽位等待和 blocking 解码等待，并让 permit 持有到任务结束。

server-rs/crates/api-server/src/raw_image.rs

@@ -13,6 +13,7 @@ use platform_image::{
 use serde::Serialize;
 use serde_json::json;
 use std::io::Cursor;
+use std::time::{Duration, Instant};
 
 use crate::{
     asset_billing::{
@@ -25,7 +26,7 @@ use crate::{
         require_openai_image_settings,
     },
     request_context::RequestContext,
-    state::AppState,
+    state::{AppState, RAW_IMAGE_DECODE_MAX_CONCURRENCY},
     tracking::record_external_generation_run_after_success,
 };
 use time::OffsetDateTime;
@@ -60,6 +61,7 @@ pub(crate) struct RawImageEditResponse {
 }
 
 const RAW_IMAGE_MAX_TEXT_FIELD_BYTES: usize = 16 * 1024;
+const RAW_IMAGE_PREPARE_TIMEOUT: Duration = Duration::from_secs(30);
 
 pub(crate) async fn edit_raw_image(
     State(state): State<AppState>,
@@ -68,11 +70,47 @@ pub(crate) async fn edit_raw_image(
     multipart: Multipart,
 ) -> Result<Json<RawImageEditResponse>, AppError> {
     let payload = parse_multipart_request(multipart).await?;
-    let prepared = tokio::task::spawn_blocking(move || prepare_request(payload))
-        .await
-        .map_err(|error| {
-            AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR).with_message(error.to_string())
-        })??;
+    let local_deadline = Instant::now()
+        .checked_add(RAW_IMAGE_PREPARE_TIMEOUT)
+        .unwrap_or_else(Instant::now);
+    let processing_deadline = request_context
+        .external_call_deadline()
+        .map(|deadline| deadline.min(local_deadline))
+        .unwrap_or(local_deadline);
+    let permit = match tokio::time::timeout_at(
+        tokio::time::Instant::from_std(processing_deadline),
+        state.raw_image_decode_limiter().acquire_owned(),
+    )
+    .await
+    {
+        Ok(Ok(permit)) => permit,
+        Ok(Err(error)) => {
+            return Err(
+                AppError::from_status(StatusCode::SERVICE_UNAVAILABLE).with_details(json!({
+                    "provider": "raw-image-edit",
+                    "code": "RAW_IMAGE_DECODE_LIMITER_UNAVAILABLE",
+                    "message": format!("raw 图片解码并发控制器不可用：{error}"),
+                })),
+            );
+        }
+        Err(_) => return Err(raw_image_prepare_timeout_error("等待 raw 图片解码槽位超时")),
+    };
+    let worker = tokio::task::spawn_blocking(move || {
+        // 超时只能停止 async 等待，permit 必须由 blocking closure 持有到解码真正结束。
+        let _permit = permit;
+        prepare_request(payload)
+    });
+    let prepared =
+        match tokio::time::timeout_at(tokio::time::Instant::from_std(processing_deadline), worker)
+            .await
+        {
+            Ok(Ok(result)) => result?,
+            Ok(Err(error)) => {
+                return Err(AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR)
+                    .with_message(error.to_string()));
+            }
+            Err(_) => return Err(raw_image_prepare_timeout_error("raw 图片解码处理超时")),
+        };
     let settings = require_openai_image_settings(&state)?.with_external_api_audit_context(
         &request_context,
         Some(authenticated.claims().user_id().to_string()),
@@ -472,6 +510,15 @@ fn bad_request(message: impl Into<String>) -> AppError {
     }))
 }
 
+fn raw_image_prepare_timeout_error(message: &str) -> AppError {
+    AppError::from_status(StatusCode::GATEWAY_TIMEOUT).with_details(json!({
+        "provider": "raw-image-edit",
+        "code": "RAW_IMAGE_PREPARE_TIMEOUT",
+        "message": message,
+        "maxConcurrency": RAW_IMAGE_DECODE_MAX_CONCURRENCY,
+    }))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

server-rs/crates/api-server/src/state.rs

@@ -56,6 +56,8 @@ const ADMIN_ROLE: &str = "admin";
 const EDITOR_AGENT_LLM_MAX_RETRIES: u32 = 1;
 const EDITOR_AGENT_LLM_MAX_RETRY_BACKOFF_MS: u64 = 60_000;
 pub(crate) const CHARACTER_ANIMATION_OSS_MAX_CONCURRENCY: usize = 8;
+// Raw image PNG 解码会进入 Tokio blocking pool；单独限流，避免图片请求挤占其它 blocking 工作。
+pub(crate) const RAW_IMAGE_DECODE_MAX_CONCURRENCY: usize = 4;
 // P=8：父侧成功图片读取/解码槽。配 N=16 是内存与出口吞吐的折中，
 // 极端完整 body 内存按 (N + P) × 32 MiB 评估（见调度方案 §9.2）。
 pub(crate) const BGFILTER_IMAGE_VALIDATION_MAX_CONCURRENCY: usize = 8;
@@ -304,6 +306,7 @@ pub struct AppStateInner {
     matting_client: Option<MattingClient>,
     bgfilter_provider_http_client: reqwest::Client,
     bgfilter_worker_http_client: reqwest::Client,
+    raw_image_decode_limiter: Arc<Semaphore>,
     bgfilter_image_validation_limiter: Arc<Semaphore>,
     character_animation_oss_http_client: reqwest::Client,
     character_animation_oss_io_limiter: Arc<Semaphore>,
@@ -612,6 +615,9 @@ impl AppState {
         let bgfilter_image_validation_limiter = Arc::new(Semaphore::new(
             bgfilter_image_validation_concurrency.min(Semaphore::MAX_PERMITS),
         ));
+        let raw_image_decode_limiter = Arc::new(Semaphore::new(
+            RAW_IMAGE_DECODE_MAX_CONCURRENCY.min(Semaphore::MAX_PERMITS),
+        ));
         let character_animation_oss_http_client = build_character_animation_oss_http_client()?;
         let character_animation_oss_io_limiter =
             Arc::new(Semaphore::new(CHARACTER_ANIMATION_OSS_MAX_CONCURRENCY));
@@ -674,6 +680,7 @@ impl AppState {
             matting_client,
             bgfilter_provider_http_client,
             bgfilter_worker_http_client,
+            raw_image_decode_limiter,
             bgfilter_image_validation_limiter,
             character_animation_oss_http_client,
             character_animation_oss_io_limiter,
@@ -1584,6 +1591,10 @@ impl AppState {
         &self.bgfilter_worker_http_client
     }
 
+    pub fn raw_image_decode_limiter(&self) -> Arc<Semaphore> {
+        self.raw_image_decode_limiter.clone()
+    }
+
     pub fn bgfilter_worker_reached(&self) -> bool {
         self.bgfilter_worker_reached.load(Ordering::Relaxed)
     }