fix: recover stalled article content fetches

- enforce a process-level timeout for browser content fetches
- recover stale FETCHING rows and claim articles one at a time
- route manual refreshes through the same bounded fetch path
- cover timeout, recovery, migration, and batch continuation

apis/article.py

@@ -15,7 +15,6 @@ from core.print import print_warning, print_info, print_error, print_success
 from core.cache import clear_cache_pattern
 from tools.fix import fix_article
 from core.article_content import sync_article_content
-from driver.wxarticle import WXArticleFetcher
 router = APIRouter(prefix=f"/articles", tags=["文章管理"])
 
 _refresh_tasks = {}
@@ -38,13 +37,11 @@ def _get_active_refresh_task(article_id: str):
 
 
 def _run_refresh_article_task_wrapper(task_id: str, article_id: str):
-    """包装器:在线程中运行 async 函数"""
-    import asyncio
-    asyncio.run(_run_refresh_article_task(task_id, article_id))
+    _run_refresh_article_task(task_id, article_id)
 
-async def _run_refresh_article_task(task_id: str, article_id: str):
+
+def _run_refresh_article_task(task_id: str, article_id: str):
     session = DB.get_session()
-    fetcher = None
     try:
         _set_refresh_task(task_id, {
             "task_id": task_id,
@@ -63,48 +60,26 @@ async def _run_refresh_article_task(task_id: str, article_id: str):
             })
             return
 
-        target_url = (article.url or "").strip()
-        if not target_url:
+        updated, fetch_mode = sync_article_content(
+            session=session,
+            article=article,
+            preferred_mode=cfg.get("gather.content_mode", "web"),
+            force=True,
+        )
+        if not updated:
             _set_refresh_task(task_id, {
                 "task_id": task_id,
                 "article_id": article_id,
                 "status": "failed",
-                "message": "文章缺少可抓取链接"
+                "message": f"文章刷新失败: {fetch_mode}"
             })
             return
 
-        fetcher = WXArticleFetcher()
-        fetched = await fetcher.get_article_content(target_url)
-        fetched_content = fetched.get("content")
-        article.show_type=fetched.get("article_type",article.show_type )
-        if fetched_content != "DELETED" and not fetched_content:
-            fetch_error = fetched.get("fetch_error") or "文章内容抓取为空"
-            _set_refresh_task(task_id, {
-                "task_id": task_id,
-                "article_id": article_id,
-                "status": "failed",
-                "message": f"文章刷新失败: {fetch_error}"
-            })
-            return
-
-        article.title = fetched.get("title") or article.title
-        article.url = target_url
-        article.publish_time = fetched.get("publish_time") or article.publish_time
-        article.content = fetched_content if fetched_content is not None else article.content
-        if fetched_content == "DELETED":
-            article.description = fetched.get("description") or article.description
-        else:
-            article.description = fetched.get("description") or article.description
-        article.pic_url = fetched.get("topic_image") or fetched.get("pic_url") or article.pic_url
-        article.status = DATA_STATUS.DELETED if fetched_content == "DELETED" else DATA_STATUS.ACTIVE
-        # 更新 has_content 字段
-        article.has_content = 1 if (article.content and article.content.strip()) else 0
-
         now_seconds = int(time.time())
-        now_millis = int(time.time() * 1000)
         article.updated_at = now_seconds
-        article.updated_at_millis = now_millis
+        article.updated_at_millis = int(time.time() * 1000)
         session.commit()
+        session.refresh(article)
 
         clear_cache_pattern("articles_list")
         clear_cache_pattern("article_detail")
@@ -115,7 +90,12 @@ async def _run_refresh_article_task(task_id: str, article_id: str):
             "task_id": task_id,
             "article_id": article_id,
             "status": "success",
-            "message": "文章刷新成功",
+            "message": (
+                "文章已被发布者删除"
+                if article.status == DATA_STATUS.DELETED
+                else "文章刷新成功"
+            ),
+            "fetch_mode": fetch_mode,
             "updated_at": now_seconds
         })
     except Exception as e:

config.example.yaml

@@ -132,7 +132,17 @@ gather:
   content_auto_interval: ${GATHER.CONTENT_AUTO_INTERVAL:-59}
   #内容修正模式，默认web 允许值 web、api
   content_mode: ${GATHER.CONTENT_MODE:-web}
-  #是否清理html标签 默认True 
+  #单篇正文抓取总超时（秒），超时会终止浏览器进程并释放文章锁
+  content_fetch_timeout: ${GATHER.CONTENT_FETCH_TIMEOUT:-60}
+  #FETCHING锁超过该时间（秒）后自动恢复；历史无时间戳锁也会恢复
+  content_fetch_stale_timeout: ${GATHER.CONTENT_FETCH_STALE_TIMEOUT:-300}
+  #每轮补抓文章数量，避免超过内容队列的总任务超时
+  content_batch_size: ${GATHER.CONTENT_BATCH_SIZE:-5}
+  #单篇正文最大失败次数
+  content_max_failures: ${GATHER.CONTENT_MAX_FAILURES:-3}
+  #首选模式失败后是否回退到另一模式；设为False可启用api-only
+  content_fallback: ${GATHER.CONTENT_FALLBACK:-True}
+  #是否清理html标签 默认True
   clean_html: ${GATHER.CLEAN_HTML:-False}
   #浏览器类型 默认firefox 允许值 firefox/edge/webkit
   browser_type: ${BROWSER_TYPE:-firefox}

core/article_content.py

@@ -5,6 +5,11 @@ from typing import Any, Tuple
 from core.config import cfg
 from core.models.base import DATA_STATUS
 from core.print import print_info, print_warning
+from core.process_timeout import (
+    ProcessExecutionError,
+    ProcessExecutionTimeout,
+    run_in_process,
+)
 
 
 def normalize_content_mode(mode: str | None = None) -> str:
@@ -55,9 +60,17 @@ def _fetch_with_api(url: str) -> Tuple[str, Any]:
     return (fetcher.content_extract(url) or "").strip(),{}
 
 
-def fetch_article_content(url: str, preferred_mode: str | None = None) -> Tuple[str, str,str]:
+def _fetch_article_content_unbounded(
+    url: str,
+    preferred_mode: str | None = None,
+    allow_fallback: bool = True,
+) -> Tuple[str, str, str]:
     mode = normalize_content_mode(preferred_mode)
-    modes = [mode] + [item for item in ("web", "api") if item != mode]
+    modes = [mode]
+    if allow_fallback:
+        modes += [item for item in ("web", "api") if item != mode]
+
+    article_type = ""
 
     for current_mode in modes:
         try:
@@ -65,7 +78,6 @@ def fetch_article_content(url: str, preferred_mode: str | None = None) -> Tuple[
                 content,result = _fetch_with_api(url)
             else:
                 content,result = _fetch_with_web(url)
-            print(result)
             article_type = result.get("article_type", "")
         except Exception as exc:
             print_warning(f"fetch article content failed in {current_mode} mode: {exc}")
@@ -76,7 +88,44 @@ def fetch_article_content(url: str, preferred_mode: str | None = None) -> Tuple[
         if content:
             return content, current_mode,article_type
 
-    return "", mode,article_type
+    return "", mode, article_type
+
+
+def fetch_article_content(
+    url: str,
+    preferred_mode: str | None = None,
+    timeout: float | None = None,
+) -> Tuple[str, str, str]:
+    """Fetch article content in an isolated process with a hard timeout."""
+    fetch_timeout = float(timeout or cfg.get("gather.content_fetch_timeout", 60) or 60)
+    allow_fallback = bool(cfg.get("gather.content_fallback", True))
+    return run_in_process(
+        _fetch_article_content_unbounded,
+        url,
+        preferred_mode,
+        allow_fallback,
+        timeout=fetch_timeout,
+    )
+
+
+def mark_article_fetch_failed(session, article: Any, reason: str) -> None:
+    failures = int(getattr(article, "fix_fail_count", 0) or 0) + 1
+    max_failures = int(cfg.get("gather.content_max_failures", 3) or 3)
+    has_existing_content = bool((getattr(article, "content", "") or "").strip())
+    article.fix_fail_count = failures
+    article.has_content = 1 if has_existing_content else 0
+    if hasattr(article, "fetch_started_at"):
+        article.fetch_started_at = None
+    article.status = (
+        DATA_STATUS.FAILED
+        if failures >= max_failures and not has_existing_content
+        else DATA_STATUS.ACTIVE
+    )
+    session.commit()
+    print_warning(
+        f"article {getattr(article, 'id', '')} content fetch failed "
+        f"({failures}/{max_failures}): {reason}"
+    )
 
 
 def sync_article_content(
@@ -90,6 +139,9 @@ def sync_article_content(
         if getattr(article, "has_content", 0) == 0:
             print_info(f"article {article.id} already has content, skipping fetch")
             article.has_content = 1
+            article.status = DATA_STATUS.ACTIVE
+            if hasattr(article, "fetch_started_at"):
+                article.fetch_started_at = None
             session.commit()
             session.refresh(article)
             return True, "cached"
@@ -98,10 +150,24 @@ def sync_article_content(
     article_url = build_article_url(article)
     if not article_url:
         print_warning(f"article {getattr(article, 'id', '')} has no valid url")
+        mark_article_fetch_failed(session, article, "missing_url")
         return False, "missing_url"
 
-    content, mode, article_type = fetch_article_content(article_url, preferred_mode)
+    try:
+        content, mode, article_type = fetch_article_content(article_url, preferred_mode)
+    except ProcessExecutionTimeout as exc:
+        mark_article_fetch_failed(session, article, str(exc))
+        return False, "timeout"
+    except ProcessExecutionError as exc:
+        mark_article_fetch_failed(session, article, str(exc))
+        return False, "process_error"
+    except Exception as exc:
+        session.rollback()
+        mark_article_fetch_failed(session, article, str(exc))
+        return False, "error"
+
     if not content:
+        mark_article_fetch_failed(session, article, f"empty response via {mode}")
         return False, mode
 
     try:
@@ -110,6 +176,8 @@ def sync_article_content(
             article.content_html = ""
             article.status = DATA_STATUS.DELETED
             article.has_content = 0
+            if hasattr(article, "fetch_started_at"):
+                article.fetch_started_at = None
             session.commit()
             session.refresh(article)
             print_info(f"article {article.id} marked as deleted via {mode}")
@@ -123,6 +191,8 @@ def sync_article_content(
         article.show_type=article_type or article.show_type
         article.status = DATA_STATUS.ACTIVE
         article.has_content = 1
+        if hasattr(article, "fetch_started_at"):
+            article.fetch_started_at = None
         if not (getattr(article, "description", "") or "").strip():
             article.description = Web.get_description(content)
         # 修正成功,重置失败计数
@@ -132,13 +202,7 @@ def sync_article_content(
         session.refresh(article)
         print_info(f"article {article.id} content synced via {mode}")
         return True, mode
-    except Exception:
-        # 修正失败,增加失败计数
-        if hasattr(article, 'fix_fail_count'):
-            article.fix_fail_count = (article.fix_fail_count or 0) + 1
-            try:
-                session.commit()
-            except Exception:
-                session.rollback()
+    except Exception as exc:
         session.rollback()
-        raise
+        mark_article_fetch_failed(session, article, str(exc))
+        return False, "save_error"

core/db.py

@@ -94,6 +94,9 @@ class Db:
             if "has_content" not in columns:
                 alter_statements.append("ALTER TABLE articles ADD COLUMN has_content INTEGER DEFAULT 0")
 
+            if "fetch_started_at" not in columns:
+                alter_statements.append("ALTER TABLE articles ADD COLUMN fetch_started_at BIGINT")
+
             if not alter_statements:
                 return
 

core/models/article.py

@@ -40,6 +40,7 @@ class ArticleBase(Base):
     is_favorite = Column(Integer, default=0)  # 是否收藏
     fix_fail_count = Column(Integer, default=0)  # 修正内容失败次数
     has_content = Column(Integer, default=0, index=True)  # 是否有正文内容（0=无，1=有），用于加速查询
+    fetch_started_at = Column(BigInteger)  # 正文抓取认领时间（Unix毫秒），用于恢复陈旧锁
 class Article(ArticleBase):
     content = Column(Text)
     content_html = Column(Text)
@@ -82,5 +83,6 @@ class Article(ArticleBase):
             'is_read': self.is_read,
             'is_favorite': self.is_favorite,
             'fix_fail_count': self.fix_fail_count,
-            'has_content': self.has_content
+            'has_content': self.has_content,
+            'fetch_started_at': self.fetch_started_at
         }

core/process_timeout.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import multiprocessing
+import os
+import signal
+from collections.abc import Callable
+from typing import Any
+
+
+class ProcessExecutionError(RuntimeError):
+    pass
+
+
+class ProcessExecutionTimeout(TimeoutError):
+    pass
+
+
+def _process_entry(connection, target: Callable, args: tuple, kwargs: dict) -> None:
+    if os.name == "posix":
+        os.setsid()
+
+    try:
+        connection.send(("ok", target(*args, **kwargs)))
+    except BaseException as exc:
+        connection.send(("error", f"{type(exc).__name__}: {exc}"))
+    finally:
+        connection.close()
+
+
+def _terminate_process_tree(process, cleanup_timeout: float) -> None:
+    if not process.is_alive():
+        process.join(timeout=cleanup_timeout)
+        return
+
+    if os.name == "posix":
+        try:
+            os.killpg(process.pid, signal.SIGTERM)
+        except (ProcessLookupError, PermissionError):
+            if process.is_alive():
+                process.terminate()
+    else:
+        process.terminate()
+
+    process.join(timeout=cleanup_timeout)
+    if not process.is_alive():
+        return
+
+    if os.name == "posix":
+        try:
+            os.killpg(process.pid, signal.SIGKILL)
+        except (ProcessLookupError, PermissionError):
+            if process.is_alive():
+                process.kill()
+    else:
+        process.kill()
+    process.join(timeout=cleanup_timeout)
+
+
+def run_in_process(
+    target: Callable,
+    *args: Any,
+    timeout: float,
+    cleanup_timeout: float = 5.0,
+    **kwargs: Any,
+) -> Any:
+    """Run a callable with a wall-clock timeout that can kill child processes."""
+    context = multiprocessing.get_context("spawn")
+    receive_connection, send_connection = context.Pipe(duplex=False)
+    process = context.Process(
+        target=_process_entry,
+        args=(send_connection, target, args, kwargs),
+    )
+    process.start()
+    send_connection.close()
+
+    try:
+        if not receive_connection.poll(timeout):
+            _terminate_process_tree(process, cleanup_timeout)
+            raise ProcessExecutionTimeout(
+                f"process exceeded wall-clock timeout of {timeout:.1f}s"
+            )
+
+        try:
+            status, payload = receive_connection.recv()
+        except EOFError as exc:
+            raise ProcessExecutionError(
+                f"process exited without a result (exit code {process.exitcode})"
+            ) from exc
+
+        process.join(timeout=cleanup_timeout)
+        if process.is_alive():
+            _terminate_process_tree(process, cleanup_timeout)
+
+        if status == "error":
+            raise ProcessExecutionError(payload)
+        return payload
+    finally:
+        receive_connection.close()
+        if process.is_alive():
+            _terminate_process_tree(process, cleanup_timeout)

jobs/fetch_no_article.py

@@ -3,8 +3,90 @@ import core.db as db
 from core.config import cfg
 from core.wait import Wait
 from core.print import print_success,print_error,print_warning
-from core.article_content import build_article_url, sync_article_content
+from core.article_content import (
+    build_article_url,
+    mark_article_fetch_failed,
+    sync_article_content,
+)
 DB=db.Db(tag="内容修正")
+
+
+def recover_stale_fetching_articles(session, now_millis: int = None) -> int:
+    """Release expired or legacy FETCHING locks and account for the failure."""
+    import time
+    from sqlalchemy import case, func, or_
+
+    now_millis = now_millis or int(time.time() * 1000)
+    stale_seconds = int(cfg.get("gather.content_fetch_stale_timeout", 300) or 300)
+    stale_before = now_millis - stale_seconds * 1000
+    max_failures = int(cfg.get("gather.content_max_failures", 3) or 3)
+    next_fail_count = func.coalesce(Article.fix_fail_count, 0) + 1
+
+    recovered = session.query(Article).filter(
+        Article.status == DATA_STATUS.FETCHING,
+        or_(
+            Article.fetch_started_at.is_(None),
+            Article.fetch_started_at < stale_before,
+        ),
+    ).update(
+        {
+            Article.status: case(
+                (next_fail_count >= max_failures, DATA_STATUS.FAILED),
+                else_=DATA_STATUS.ACTIVE,
+            ),
+            Article.fix_fail_count: next_fail_count,
+            Article.fetch_started_at: None,
+        },
+        synchronize_session=False,
+    )
+    session.commit()
+    if recovered:
+        print_warning(f"已恢复 {recovered} 篇陈旧 FETCHING 文章")
+    return recovered
+
+
+def claim_next_article(session, excluded_ids=None, now_millis: int = None):
+    """Atomically claim one article immediately before it is fetched."""
+    import time
+    from sqlalchemy import or_
+
+    excluded_ids = excluded_ids or set()
+    max_failures = int(cfg.get("gather.content_max_failures", 3) or 3)
+
+    while True:
+        query = session.query(Article).filter(
+            Article.has_content == 0,
+            Article.status != DATA_STATUS.FETCHING,
+            Article.status != DATA_STATUS.DELETED,
+            or_(Article.fix_fail_count.is_(None), Article.fix_fail_count < max_failures),
+        )
+        if excluded_ids:
+            query = query.filter(~Article.id.in_(excluded_ids))
+
+        candidate = query.order_by(Article.publish_time.desc()).first()
+        if candidate is None:
+            return None
+
+        claimed = session.query(Article).filter(
+            Article.id == candidate.id,
+            Article.has_content == 0,
+            Article.status == candidate.status,
+            or_(Article.fix_fail_count.is_(None), Article.fix_fail_count < max_failures),
+        ).update(
+            {
+                Article.status: DATA_STATUS.FETCHING,
+                Article.fetch_started_at: now_millis or int(time.time() * 1000),
+            },
+            synchronize_session=False,
+        )
+        session.commit()
+        if claimed:
+            session.expire_all()
+            return session.query(Article).filter(Article.id == candidate.id).first()
+
+        session.expire_all()
+
+
 def fetch_articles_without_content():
     """
     查询content为空的文章，调用微信内容提取方法获取内容并更新数据库
@@ -12,32 +94,20 @@ def fetch_articles_without_content():
     """
     session = DB.get_session()
     try:
-        # 查询content为空且未被锁定的文章
-        from sqlalchemy import or_
-        articles = session.query(Article).filter(
-            or_(Article.has_content==0),
-            Article.status != DATA_STATUS.FETCHING,  # 排除正在获取的文章
-            Article.status != DATA_STATUS.DELETED,  # 已删除文章不再参与自动补抓
-            or_(Article.fix_fail_count.is_(None), Article.fix_fail_count < 3)  # 排除失败3次及以上的文章
-        ).order_by(Article.publish_time.desc()).limit(10).all()
-        
-        if not articles:
-            print_warning("暂无需要获取内容的文章")
-            return
+        recover_stale_fetching_articles(session)
+        batch_size = int(cfg.get("gather.content_batch_size", 5) or 5)
+        processed_ids = set()
 
-        original_status_map = {
-            article.id: article.status for article in articles
-        }
-        
-        # 锁定文章状态，防止其他节点获取
-        article_ids = [a.id for a in articles]
-        session.query(Article).filter(Article.id.in_(article_ids)).update(
-            {Article.status: DATA_STATUS.FETCHING},
-            synchronize_session=False
-        )
-        session.commit()
-        
-        for article in articles:
+        for _ in range(batch_size):
+            article = claim_next_article(session, excluded_ids=processed_ids)
+            if article is None:
+                if not processed_ids:
+                    print_warning("暂无需要获取内容的文章")
+                break
+
+            processed_ids.add(article.id)
+            article_id = article.id
+            article_title = article.title
             try:
                 url = build_article_url(article)
                 print(f"正在处理文章: {article.title}, URL: {url}")
@@ -53,20 +123,15 @@ def fetch_articles_without_content():
                         print_error(f"获取文章 {article.title} 内容已被发布者删除")
                     else:
                         print_success(f"成功更新文章 {article.title} 的内容, mode={fetch_mode} url: http://127.0.0.1:{cfg.get('port', 8001)}/views/article/{article.id}")
-                        # 成功获取内容后，恢复原始状态（通常为 ACTIVE），释放 FETCHING 锁
-                        article.status = original_status_map.get(article.id, DATA_STATUS.ACTIVE)
-                        session.commit()
                 else:
-                    # 获取失败，恢复状态以便后续重试
-                    article.status = original_status_map.get(article.id, DATA_STATUS.ACTIVE)
-                    session.commit()
                     print_error(f"获取文章 {article.title} 内容失败, mode={fetch_mode}")
                 Wait(min=5,max=10,tips=f"修正 {article.title}... 完成")
             except Exception as e:
-                # 单篇文章处理失败，恢复状态
-                article.status = original_status_map.get(article.id, DATA_STATUS.ACTIVE)
-                session.commit()
-                print_error(f"处理文章 {article.title} 时发生错误: {e}")
+                session.rollback()
+                article = session.query(Article).filter(Article.id == article_id).first()
+                if article and article.status == DATA_STATUS.FETCHING:
+                    mark_article_fetch_failed(session, article, str(e))
+                print_error(f"处理文章 {article_title} 时发生错误: {e}")
     except Exception as e:
         print_error(f"处理过程中发生错误: {e}")
         raise  # 重新抛出异常，让队列记录错误

tests/test_article_refresh_task.py

@@ -0,0 +1,58 @@
+import unittest
+from unittest.mock import patch
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from core.config import cfg
+
+cfg.config["db"] = "sqlite:////tmp/werss-refresh-task-import.db"
+
+from apis import article as article_api
+from core.models.article import Article
+from core.models.base import Base, DATA_STATUS
+
+
+class ArticleRefreshTaskTest(unittest.TestCase):
+    def setUp(self):
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(self.engine)
+        self.Session = sessionmaker(bind=self.engine)
+        self.session = self.Session()
+        self.session.add(
+            Article(
+                id="article-id",
+                title="article",
+                url="https://mp.weixin.qq.com/s/article-id",
+                content="",
+                has_content=0,
+                status=DATA_STATUS.ACTIVE,
+            )
+        )
+        self.session.commit()
+        article_api._refresh_tasks.clear()
+
+    def tearDown(self):
+        self.session.close()
+        self.engine.dispose()
+
+    def test_manual_refresh_uses_shared_sync_and_reports_timeout(self):
+        with (
+            patch.object(article_api.DB, "get_session", return_value=self.session),
+            patch.object(
+                article_api,
+                "sync_article_content",
+                return_value=(False, "timeout"),
+            ) as sync_content,
+        ):
+            article_api._run_refresh_article_task("task-id", "article-id")
+
+        sync_content.assert_called_once()
+        self.assertTrue(sync_content.call_args.kwargs["force"])
+        task = article_api._refresh_tasks["task-id"]
+        self.assertEqual(task["status"], "failed")
+        self.assertIn("timeout", task["message"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_content_fetch_queue.py

@@ -0,0 +1,217 @@
+import importlib.util
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from sqlalchemy import create_engine, inspect, text
+from sqlalchemy.orm import sessionmaker
+
+from core.config import cfg
+
+cfg.config["db"] = "sqlite:////tmp/werss-content-queue-import.db"
+
+from core import article_content
+from core.models.article import Article
+from core.models.base import Base, DATA_STATUS
+
+module_spec = importlib.util.spec_from_file_location(
+    "fetch_no_article_under_test",
+    Path(__file__).parents[1] / "jobs" / "fetch_no_article.py",
+)
+fetch_no_article = importlib.util.module_from_spec(module_spec)
+module_spec.loader.exec_module(fetch_no_article)
+
+
+def config_value(key, default=None):
+    values = {
+        "gather.content_batch_size": 2,
+        "gather.content_fetch_stale_timeout": 300,
+        "gather.content_max_failures": 3,
+        "gather.content_mode": "web",
+    }
+    return values.get(key, default)
+
+
+class ContentFetchQueueTest(unittest.TestCase):
+    def setUp(self):
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(self.engine)
+        self.Session = sessionmaker(bind=self.engine)
+        self.session = self.Session()
+        self.config_patch = patch.object(cfg, "get", side_effect=config_value)
+        self.config_patch.start()
+
+    def tearDown(self):
+        self.config_patch.stop()
+        self.session.close()
+        self.engine.dispose()
+
+    def add_article(self, article_id, **values):
+        defaults = {
+            "id": article_id,
+            "title": article_id,
+            "url": f"https://mp.weixin.qq.com/s/{article_id}",
+            "publish_time": 1,
+            "status": DATA_STATUS.ACTIVE,
+            "has_content": 0,
+            "fix_fail_count": 0,
+        }
+        defaults.update(values)
+        article = Article(**defaults)
+        self.session.add(article)
+        self.session.commit()
+        return article
+
+    def test_recovers_legacy_and_stale_locks_but_not_fresh_lock(self):
+        now_millis = 1_000_000
+        self.add_article(
+            "legacy",
+            status=DATA_STATUS.FETCHING,
+            fetch_started_at=None,
+        )
+        self.add_article(
+            "stale",
+            status=DATA_STATUS.FETCHING,
+            fetch_started_at=600_000,
+        )
+        self.add_article(
+            "fresh",
+            status=DATA_STATUS.FETCHING,
+            fetch_started_at=900_000,
+        )
+
+        recovered = fetch_no_article.recover_stale_fetching_articles(
+            self.session,
+            now_millis=now_millis,
+        )
+
+        self.assertEqual(recovered, 2)
+        for article_id in ("legacy", "stale"):
+            article = self.session.get(Article, article_id)
+            self.assertEqual(article.status, DATA_STATUS.ACTIVE)
+            self.assertEqual(article.fix_fail_count, 1)
+            self.assertIsNone(article.fetch_started_at)
+        fresh = self.session.get(Article, "fresh")
+        self.assertEqual(fresh.status, DATA_STATUS.FETCHING)
+        self.assertEqual(fresh.fix_fail_count, 0)
+        self.assertEqual(fresh.fetch_started_at, 900_000)
+
+    def test_claims_only_one_article(self):
+        self.add_article("older", publish_time=1)
+        self.add_article("newer", publish_time=2)
+
+        claimed = fetch_no_article.claim_next_article(
+            self.session,
+            now_millis=123_000,
+        )
+
+        self.assertEqual(claimed.id, "newer")
+        self.assertEqual(claimed.status, DATA_STATUS.FETCHING)
+        self.assertEqual(claimed.fetch_started_at, 123_000)
+        self.assertEqual(self.session.get(Article, "older").status, DATA_STATUS.ACTIVE)
+
+    def test_legacy_table_gets_fetch_started_at_column(self):
+        engine = create_engine("sqlite:///:memory:")
+        with engine.begin() as connection:
+            connection.execute(text("CREATE TABLE articles (id VARCHAR(255) PRIMARY KEY)"))
+
+        database = fetch_no_article.db.Db.__new__(fetch_no_article.db.Db)
+        database.engine = engine
+        database.tag = "test"
+        database.ensure_article_columns()
+
+        columns = {column["name"] for column in inspect(engine).get_columns("articles")}
+        self.assertIn("fetch_started_at", columns)
+        engine.dispose()
+
+    def test_empty_fetch_increments_failure_and_releases_lock(self):
+        article = self.add_article(
+            "empty",
+            status=DATA_STATUS.FETCHING,
+            fetch_started_at=123_000,
+        )
+
+        with patch.object(
+            article_content,
+            "fetch_article_content",
+            return_value=("", "web", ""),
+        ):
+            updated, mode = article_content.sync_article_content(
+                self.session,
+                article,
+            )
+
+        self.assertFalse(updated)
+        self.assertEqual(mode, "web")
+        self.assertEqual(article.fix_fail_count, 1)
+        self.assertEqual(article.status, DATA_STATUS.ACTIVE)
+        self.assertIsNone(article.fetch_started_at)
+
+    def test_failed_forced_refresh_preserves_existing_content(self):
+        article = self.add_article(
+            "existing",
+            content="<p>existing body</p>",
+            has_content=1,
+        )
+
+        with patch.object(
+            article_content,
+            "fetch_article_content",
+            return_value=("", "web", ""),
+        ):
+            updated, _ = article_content.sync_article_content(
+                self.session,
+                article,
+                force=True,
+            )
+
+        self.assertFalse(updated)
+        self.assertEqual(article.content, "<p>existing body</p>")
+        self.assertEqual(article.has_content, 1)
+        self.assertEqual(article.status, DATA_STATUS.ACTIVE)
+
+    def test_api_only_mode_does_not_fall_back_to_web(self):
+        with (
+            patch.object(article_content, "_fetch_with_api", return_value=("", {})),
+            patch.object(article_content, "_fetch_with_web") as web_fetch,
+        ):
+            content, mode, _ = article_content._fetch_article_content_unbounded(
+                "https://example.com/article",
+                preferred_mode="api",
+                allow_fallback=False,
+            )
+
+        self.assertEqual(content, "")
+        self.assertEqual(mode, "api")
+        web_fetch.assert_not_called()
+
+    def test_failed_article_does_not_prevent_next_batch_item(self):
+        self.add_article("first", publish_time=2)
+        self.add_article("second", publish_time=1)
+        processed_ids = []
+
+        def fail_fetch(session, article, preferred_mode=None):
+            processed_ids.append(article.id)
+            article.status = DATA_STATUS.ACTIVE
+            article.fetch_started_at = None
+            article.fix_fail_count += 1
+            session.commit()
+            return False, "web"
+
+        with (
+            patch.object(
+                fetch_no_article,
+                "DB",
+                SimpleNamespace(get_session=self.Session),
+            ),
+            patch.object(fetch_no_article, "sync_article_content", side_effect=fail_fetch),
+            patch.object(fetch_no_article, "Wait"),
+        ):
+            fetch_no_article.fetch_articles_without_content()
+
+        self.assertEqual(processed_ids, ["first", "second"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_process_timeout.py

@@ -0,0 +1,46 @@
+import time
+import unittest
+
+from core.process_timeout import (
+    ProcessExecutionError,
+    ProcessExecutionTimeout,
+    run_in_process,
+)
+
+
+def return_value(value):
+    return value
+
+
+def sleep_for(seconds):
+    time.sleep(seconds)
+
+
+def raise_remote_error():
+    raise ValueError("remote failure")
+
+
+class ProcessTimeoutTest(unittest.TestCase):
+    def test_returns_successful_result(self):
+        self.assertEqual(run_in_process(return_value, "ok", timeout=2), "ok")
+
+    def test_terminates_hung_process(self):
+        started_at = time.monotonic()
+
+        with self.assertRaises(ProcessExecutionTimeout):
+            run_in_process(
+                sleep_for,
+                10,
+                timeout=0.2,
+                cleanup_timeout=0.5,
+            )
+
+        self.assertLess(time.monotonic() - started_at, 2)
+
+    def test_surfaces_remote_exception(self):
+        with self.assertRaisesRegex(ProcessExecutionError, "remote failure"):
+            run_in_process(raise_remote_error, timeout=2)
+
+
+if __name__ == "__main__":
+    unittest.main()