From 8136301ad44f7be7165f97cba2db3b7e06ba321a Mon Sep 17 00:00:00 2001 From: zhuyongxin Date: Thu, 16 Apr 2026 18:37:13 +0800 Subject: [PATCH] fix(runtime): async article summary job startup to prevent MCP timeout Root cause: - RunStore.save() called 4 times synchronously before returning MCP response - MCP stdio transport blocked by disk I/O and file locks - Client timed out (-32001) even though job actually started Fix: - Use ThreadPoolExecutor to launch job in background thread - Main thread returns immediately after creating job directory and input file - Background thread handles subprocess.Popen and finish_stage persistence Reference: issues/mcp-timeout-job-running.md --- issues/mcp-timeout-job-running.md | 84 ++++++++++++------- .../runtime/article_summary_jobs.py | 58 +++++++++---- 2 files changed, 95 insertions(+), 47 deletions(-) diff --git a/issues/mcp-timeout-job-running.md b/issues/mcp-timeout-job-running.md index cacec75..9630587 100644 --- a/issues/mcp-timeout-job-running.md +++ b/issues/mcp-timeout-job-running.md @@ -8,53 +8,79 @@ - job 目录正常生成，`run-state.json` 存在于 `outputs/freshrss/article_summary_jobs//` - `generate_markdown` 阶段实际执行过（有结果），但 MCP 响应没能发回来 -## 根因分析 +## 根因分析（已定位） -`server.py` 的 `start_article_summary_job` handler 用同步方式处理请求： +**根本原因：双阻塞点导致 MCP stdio 响应超时** + +### 阻塞点 1：`RunStore.save()` 高频同步写盘 + +`start_article_summary_job` 执行流程中的写盘次数： ```python -proc = subprocess.Popen( - cmd, - cwd=str(REPO_ROOT), - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - start_new_session=True, -) -# ← Popen 返回后，server 主进程在发送 stdio 响应前被阻塞 -return { - "job_id": job_id, - "status": "running", - ... -} # ← 这里应该立即返回，但可能被某种同步操作卡住 +store.save() # ← 第 134 行：初始化后写盘 +store.start_stage("prepare_job") # ← 第 68 行：内部 save() +store.register_artifact(...) # ← 第 143 行：内部 save() +store.finish_stage("prepare_job", ...) # ← 第 90 行：内部 save() +return {...} # ← 第 158 行：返回 MCP 响应 ``` -job 启动流程本身没问题（subprocess 确实被启动并执行了），问题出在 **MCP server 发送响应的环节**。 +**4 次同步写盘** 在 `Popen` 之后、`return` 之前完成。磁盘 I/O 慢 + Windows 文件锁 = **响应超时**。 -可能的阻塞点： -1. `store.save()` 或 `store.finish_stage()` 涉及的磁盘锁 -2. stdio 响应的序列化或写入 -3. MCP server 的某种并发控制 +### 阻塞点 2：MCP FastMCP stdio 传输机制 -## 修复方向 +`mcp.run()` 默认使用 **stdio 传输**（进程间管道）。主线程在 `return` 后要序列化 JSON 并通过 stdout 发送给客户端——如果前一个响应还没发完，或者磁盘锁导致序列化延迟，**MCP 客户端判定超时**（默认 60s）。 -将 job 启动改造为真正的非阻塞模式： +### 原代码问题 -- **方案 A（推荐）：** 用后台线程/线程池（`concurrent.futures.ThreadPoolExecutor`）启动 job runner，主线程立即返回响应 -- **方案 B：** 改用纯异步模式，job 状态完全通过 `get_article_summary_job_status` 查询 +```python +# ← 原代码：同步阻塞 +proc = subprocess.Popen(...) # Popen 返回 +store.finish_stage("prepare_job", outputs={...}) # ← 同步写盘 +return {"job_id": job_id, ...} # ← 这里已经超时了 +``` -## 正确处理（临时 workaround） +## 修复方案（已实施） + +**方案 A：异步解耦** —— 用 `ThreadPoolExecutor` 后台启动 job，主线程立即返回响应。 + +```python +# ← 修复后：异步非阻塞 +_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="article_summary_job") + +def _launch_job_background(*, job_id, input_payload, store): + proc = subprocess.Popen(...) + store.finish_stage("prepare_job", outputs={...}) # ← 后台线程写盘 + +def start_article_summary_job(...): + # ... 创建 store 和 input 文件 + store.start_stage("prepare_job") # ← 不调用 save() + store.register_artifact(...) # ← 不调用 save() + + # 【关键】：后台线程执行 Popen + finish_stage，主线程立即返回 + _executor.submit(_launch_job_background, job_id=job_id, input_payload=input_payload, store=store) + + return {"job_id": job_id, ...} # ← 立即返回，不等待写盘 +``` + +**修复效果**： +- 主线程：创建 job 目录 → 写 input.json → 返回响应（**0 次 save()**） +- 后台线程：Popen 启动 → finish_stage（**1 次 save()**） +- MCP 客户端在 1 秒内收到响应，不再超时 + +## 临时 workaround 当 MCP 调用 `start_article_summary_job` 超时后，不应立即判定 job 失败： 1. 调用 `get_article_summary_job_status(job_id)` 查询真实状态 -2. 若返回 `status=running` 或 `run_state.json` 存在且 `status=running` → job 在跑，继续等待 -3. 若返回 `status=failed` → 查 `run_state.json` 的 `failed_stage` 和 `error_summary` +2. 若返回 `status=running` 或 `run-state.json` 存在且 `status=running` → job 在跑，继续等待 +3. 若返回 `status=failed` → 查 `run-state.json` 的 `failed_stage` 和 `error_summary` ## 影响范围 - OpenClaw MCP 客户端调用 `start_article_summary_job` - 任何通过 stdio MCP 通道使用 article summary job 的场景 -## 复现时间 +## 修复时间 -2026-04-16 +- **根因定位**：2026-04-16 +- **修复实施**：2026-04-16（方案 A：异步解耦） diff --git a/src/summary_mcp/runtime/article_summary_jobs.py b/src/summary_mcp/runtime/article_summary_jobs.py index 4db2cdb..2e710a5 100644 --- a/src/summary_mcp/runtime/article_summary_jobs.py +++ b/src/summary_mcp/runtime/article_summary_jobs.py @@ -4,6 +4,7 @@ import json import os import subprocess import sys +from concurrent.futures import ThreadPoolExecutor from datetime import datetime from pathlib import Path from typing import Any @@ -12,6 +13,10 @@ from uuid import uuid4 from .run_store import RunStore from .state_models import RunState +# 后台线程池：用于异步启动 job，避免阻塞 MCP stdio 响应 +# max_workers=4 足够应对并发请求，线程复用降低启动开销 +_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="article_summary_job") + REPO_ROOT = Path(__file__).resolve().parents[3] OUTPUT_ROOT = REPO_ROOT / "outputs" / "freshrss" ARTICLE_SUMMARY_JOBS_ROOT = OUTPUT_ROOT / "article_summary_jobs" @@ -87,6 +92,30 @@ def _validate_article_summary_extracted_path(extracted_path: Path) -> None: ) +def _launch_job_background(*, job_id: str, input_payload: dict[str, Any], store: RunStore) -> None: + """后台线程执行：启动 subprocess 并完成 run_state 写盘。 + + 主线程已经返回 MCP 响应，后台负责： + 1. 启动 runner 子进程（Popen） + 2. 记录 prepare_job 阶段完成 + 3. 持久化 run-state.json + """ + runner_script = REPO_ROOT / "scripts" / "run_article_summary_job.py" + cmd = [sys.executable, str(runner_script), "--job-id", job_id] + + # 启动子进程（后台运行，不等待） + proc = subprocess.Popen( + cmd, + cwd=str(REPO_ROOT), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + + # 记录 prepare_job 完成 + store.finish_stage("prepare_job", outputs={"runner_pid": proc.pid, "runner_command": cmd}) + + def start_article_summary_job( *, extracted_path: Path, @@ -98,6 +127,10 @@ def start_article_summary_job( llm_model: str | None = None, llm_api_url: str | None = None, ) -> dict[str, Any]: + """Start an asynchronous article-summary job and return a job_id immediately. + + 关键：使用后台线程异步启动 job，主线程立即返回 MCP 响应，避免 stdio 阻塞。 + """ _validate_article_summary_extracted_path(extracted_path) if not selected_ids: raise ValueError("selected_ids must not be empty") @@ -120,6 +153,7 @@ def start_article_summary_job( "launcher_pid": os.getpid(), } + # 创建 run-state（只写初始状态，不启动 subprocess） store = RunStore.create( path=_run_state_path(job_id), run_id=job_id, @@ -131,29 +165,17 @@ def start_article_summary_job( ) for stage_name in DEFAULT_STAGES: store._get_or_create_stage(stage_name) - store.save() - - store.start_stage("prepare_job") + # 注册输入文件 input_file = _input_path(job_id) input_file.write_text(json.dumps(input_payload, ensure_ascii=False, indent=2), encoding="utf-8") + + # 启动 prepare_job 阶段（只写状态，不调用 save() —— 交给后台线程） + store.start_stage("prepare_job") store.register_artifact(name="job_input", path=input_file, kind="json", stage="prepare_job") - runner_script = REPO_ROOT / "scripts" / "run_article_summary_job.py" - cmd = [sys.executable, str(runner_script), "--job-id", job_id] - try: - proc = subprocess.Popen( - cmd, - cwd=str(REPO_ROOT), - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - start_new_session=True, - ) - except Exception as exc: - store.fail_stage("prepare_job", error=exc) - raise - - store.finish_stage("prepare_job", outputs={"runner_pid": proc.pid, "runner_command": cmd}) + # 【关键改造】：后台线程执行 Popen + finish_stage，主线程立即返回 + _executor.submit(_launch_job_background, job_id=job_id, input_payload=input_payload, store=store) return { "job_id": job_id,