fix(plugin-market): bound catalog fetch concurrency

app/application/plugin/catalog.py

@@ -29,6 +29,7 @@ AsyncMarketLoader = Callable[
 ]
 PluginMapper = Callable[[str, dict, str, list[str], int, Optional[str]], Any]
 ProgressCallback = Callable[..., Any]
+DEFAULT_MARKET_FETCH_CONCURRENCY = 5
 
 
 @dataclass(frozen=True, slots=True)
@@ -103,8 +104,11 @@ class PluginCatalogService:
             version_compare: Callable[[str, str, str], bool],
             warning: Callable[[str], Any],
             error: Callable[[str], Any],
+            max_concurrency: int = DEFAULT_MARKET_FETCH_CONCURRENCY,
     ) -> None:
         """保存市场读取、插件映射和版本比较端口。"""
+        if max_concurrency < 1:
+            raise ValueError("插件市场读取并发必须大于 0")
         self._market_loader = market_loader
         self._async_market_loader = async_market_loader
         self._installed_plugins_provider = installed_plugins_provider
@@ -113,6 +117,7 @@ class PluginCatalogService:
         self._version_compare = version_compare
         self._warning = warning
         self._error = error
+        self._max_concurrency = max_concurrency
 
     @staticmethod
     def _load_request(
@@ -192,7 +197,9 @@ class PluginCatalogService:
             loader: Callable[[str, Optional[str], bool], list[Any]],
     ) -> list[Any]:
         """并发读取多个市场和代际，并按稳定优先级合并。"""
-        with concurrent.futures.ThreadPoolExecutor() as executor:
+        with concurrent.futures.ThreadPoolExecutor(
+            max_workers=self._max_concurrency
+        ) as executor:
             futures_meta: dict[
                 concurrent.futures.Future,
                 tuple[int, bool, int],
@@ -210,7 +217,11 @@ class PluginCatalogService:
 
             collected = []
             for future in concurrent.futures.as_completed(futures_meta):
-                plugins = future.result()
+                try:
+                    plugins = future.result()
+                except Exception as error:  # noqa: BLE001 - 单个市场失败不能中断全量刷新
+                    self._error(f"获取插件市场数据失败：{str(error)}")
+                    plugins = []
                 market_index, is_higher, flag_priority = futures_meta[future]
                 collected.append((
                     market_index,
@@ -240,6 +251,8 @@ class PluginCatalogService:
             progress_callback: Optional[ProgressCallback] = None,
     ) -> list[Any]:
         """异步读取多个市场和代际，并按调用方需要合并或保留仓库候选。"""
+        semaphore = asyncio.Semaphore(self._max_concurrency)
+
         async def fetch(
                 market: str,
                 package_version: Optional[str],
@@ -247,7 +260,8 @@ class PluginCatalogService:
                 task_index: int,
         ) -> tuple[int, str, list[Any]]:
             """读取一个市场代际并保留创建时的稳定任务序号。"""
-            plugins = await loader(market, package_version, force)
+            async with semaphore:
+                plugins = await loader(market, package_version, force)
             return task_index, result_version, plugins or []
 
         tasks: list[asyncio.Task[tuple[int, str, list[Any]]]] = []

tests/test_plugin_catalog_concurrency.py

@@ -0,0 +1,114 @@
+"""插件市场目录读取并发控制测试。"""
+
+import asyncio
+import threading
+import time
+from collections.abc import Callable
+
+import pytest
+
+from app.application.plugin.catalog import PluginCatalogService
+
+
+def _catalog(
+        max_concurrency: int = 2,
+        error: Callable[[str], None] | None = None,
+) -> PluginCatalogService:
+    """构造不依赖真实插件目录和网络的目录服务。"""
+    return PluginCatalogService(
+        market_loader=lambda *_args: {},
+        async_market_loader=lambda *_args: {},
+        installed_plugins_provider=lambda: [],
+        plugin_mapper=lambda *_args, **_kwargs: None,
+        is_local_repo=lambda _repo_url: False,
+        version_compare=lambda *_args: False,
+        warning=lambda *_args: None,
+        error=error if error is not None else lambda *_args: None,
+        max_concurrency=max_concurrency,
+    )
+
+
+def test_sync_collect_limits_market_fetch_concurrency() -> None:
+    """同步目录刷新不能因市场数量增长而创建无限线程。"""
+    lock = threading.Lock()
+    active = 0
+    peak = 0
+    calls: list[tuple[str, str | None, bool]] = []
+
+    def loader(market: str, package_version: str | None, force: bool):
+        nonlocal active, peak
+        with lock:
+            active += 1
+            peak = max(peak, active)
+            calls.append((market, package_version, force))
+        time.sleep(0.01)
+        with lock:
+            active -= 1
+        return []
+
+    result = _catalog(max_concurrency=2).collect(
+        markets=["market-a", "market-b", "market-c"],
+        compatible_flags=["v3"],
+        force=True,
+        loader=loader,
+    )
+
+    assert result == []
+    assert len(calls) == 6
+    assert peak <= 2
+
+
+def test_sync_collect_isolates_market_failure() -> None:
+    """同步目录刷新遇到单个市场异常时仍保留其他市场结果。"""
+    errors: list[str] = []
+
+    def loader(
+            market: str,
+            package_version: str | None,
+            _force: bool,
+    ):
+        if market == "market-b" and package_version == "v3":
+            raise RuntimeError("market unavailable")
+        return []
+
+    result = _catalog(max_concurrency=2, error=errors.append).collect(
+        markets=["market-a", "market-b"],
+        compatible_flags=["v3"],
+        force=True,
+        loader=loader,
+    )
+
+    assert result == []
+    assert len(errors) == 1
+    assert "market unavailable" in errors[0]
+
+
+@pytest.mark.asyncio
+async def test_async_collect_limits_market_fetch_concurrency() -> None:
+    """异步目录刷新必须用信号量限制同时在途的市场请求。"""
+    active = 0
+    peak = 0
+
+    async def loader(_market: str, _package_version: str | None, _force: bool):
+        nonlocal active, peak
+        active += 1
+        peak = max(peak, active)
+        await asyncio.sleep(0.01)
+        active -= 1
+        return []
+
+    result = await _catalog(max_concurrency=2).async_collect(
+        markets=["market-a", "market-b", "market-c"],
+        compatible_flags=["v3"],
+        force=True,
+        loader=loader,
+    )
+
+    assert result == []
+    assert peak <= 2
+
+
+def test_market_fetch_concurrency_must_be_positive() -> None:
+    """无效并发值应在服务构造时快速失败。"""
+    with pytest.raises(ValueError, match="大于 0"):
+        _catalog(max_concurrency=0)