nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-01 12:56:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #6617 from ranzhigg/fix/plugin-market-global-flight

Browse Source
This commit is contained in:
jxxghp
2026-09-09 08:21:17 +08:00
committed by GitHub
parent 7a1941ec39 0c1c4adebe
commit fb48751745
8 changed files with 337 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

184
app/adapters/external/plugin/client.py vendored
View File

@@ -9,9 +9,10 @@ import threading
import time
import traceback
from collections import Counter
from concurrent.futures import Future
from pathlib import Path
from typing import Any, Dict, Iterator, List, Optional, cast
from urllib.parse import urlparse, urlsplit
from urllib.parse import urlsplit
import httpx2
from packaging.specifiers import InvalidSpecifier, SpecifierSet
@@ -25,9 +26,11 @@ from app.domain.plugin import (
is_local_plugin_source,
is_physical_plugin_id,
is_plugin_generation_compatible,
normalize_plugin_market_repo_url as _normalize_plugin_market_repo_url,
parse_local_plugin_generation,
parse_local_plugin_path,
parse_local_plugin_reference,
split_plugin_market_repo_urls as _split_plugin_market_repo_urls,
)
from app.foundation.environment import is_free_threaded_runtime
from app.foundation.singleton import WeakSingleton
@@ -51,6 +54,7 @@ PLUGIN_INDEX_MAX_HISTORY_ENTRIES = 512
PLUGIN_INDEX_MAX_NESTING = 64
PLUGIN_INDEX_READ_CHUNK_SIZE = 64 * 1024
PLUGIN_INDEX_REQUEST_TIMEOUT = 15
DEFAULT_PLUGIN_INDEX_FETCH_CONCURRENCY = 5
PLUGIN_INDEX_COMPATIBILITY_FLAG_PATTERN = re.compile(r"^v\d+t?$")
PLUGIN_INDEX_TEXT_LIMITS = {
"name": 256,
@@ -64,6 +68,12 @@ PLUGIN_INDEX_TEXT_LIMITS = {
"repo_url": 2048,
}
# 目录服务和来源库存可能在同一进程内同时刷新；调用方各自创建的信号量
# 只能限制单次刷新，不能限制真实出站请求总量。索引传输层统一持有闸门。
_PLUGIN_INDEX_FETCH_GATE = threading.BoundedSemaphore(
DEFAULT_PLUGIN_INDEX_FETCH_CONCURRENCY
)
class _PluginIndexTooLargeError(RuntimeError):
"""插件索引的声明长度或实际读取字节超过资源边界。"""
@@ -139,35 +149,9 @@ VERSION_BACKWARD_COMPATIBLE_FLAGS: Dict[str, List[str]] = {
"v3": ["v2"],
}
def normalize_plugin_market_repo_url(repo_url: str) -> Optional[str]:
"""规范化插件仓库地址，便于跨来源合并去重。"""
repo_url = (repo_url or "").strip().rstrip("/")
if not repo_url:
return None
repo_url = repo_url.removesuffix(".git")
parsed_url = urlparse(repo_url)
if parsed_url.scheme not in {"http", "https"}:
return None
if (parsed_url.hostname or "").lower() != "github.com":
return None
paths = [item for item in parsed_url.path.split("/") if item]
if len(paths) < 2:
return None
return f"https://github.com/{paths[0]}/{paths[1]}"
def split_plugin_market_repo_urls(value: Optional[str]) -> list[str]:
"""拆分插件市场仓库配置并保持原有顺序去重。"""
repos: list[str] = []
seen_repos = set()
for item in re.split(r"[\n,，]+", value or ""):
normalized_repo = normalize_plugin_market_repo_url(item)
if not normalized_repo or normalized_repo.lower() in seen_repos:
continue
repos.append(normalized_repo)
seen_repos.add(normalized_repo.lower())
return repos
# 保留历史兼容模块的导出路径；规范化规则的唯一实现位于纯领域模块。
normalize_plugin_market_repo_url = _normalize_plugin_market_repo_url
split_plugin_market_repo_urls = _split_plugin_market_repo_urls
def extract_plugin_market_repos_from_wiki(
markdown: str, require_markers: bool = False
@@ -220,6 +204,13 @@ class PluginMarketTransport(metaclass=WeakSingleton):
"""负责插件市场、本地仓库和 GitHub 元数据读取。"""
_base_url = "https://raw.githubusercontent.com/{user}/{repo}/main/"
_index_future_lock = threading.Lock()
_index_futures: dict[tuple[str, str], Future[Optional[PluginIndex]]] = {}
_index_task_lock = threading.Lock()
_index_tasks: dict[
tuple[asyncio.AbstractEventLoop, str, str],
asyncio.Task[Optional[PluginIndex]],
] = {}
_release_task_lock = threading.Lock()
_release_tasks: dict[
tuple[asyncio.AbstractEventLoop, str, bool],
@@ -891,6 +882,8 @@ class PluginMarketTransport(metaclass=WeakSingleton):
if not repo_url:
return None
repo_url = normalize_plugin_market_repo_url(repo_url) or repo_url
user, repo = cls.get_repo_info(repo_url)
if not user or not repo:
return None
@@ -1137,6 +1130,81 @@ class PluginMarketTransport(metaclass=WeakSingleton):
releases.extend(cls.__normalize_plugin_release_response(payload))
return len(payload) >= 100
@staticmethod
def _plugin_index_key(
repo_url: str,
package_version: Optional[str],
) -> tuple[str, str]:
"""生成跨 URL 表示形式稳定的插件索引请求键。"""
normalized_repo = normalize_plugin_market_repo_url(repo_url)
return (
(normalized_repo or str(repo_url).strip().rstrip("/")).lower(),
package_version or "",
)
@classmethod
def _remove_index_future(
cls,
key: tuple[str, str],
future: Future[Optional[PluginIndex]],
) -> None:
"""同步索引请求结束后释放 single-flight 占位。"""
with cls._index_future_lock:
if cls._index_futures.get(key) is future:
cls._index_futures.pop(key, None)
@classmethod
def _remove_index_task(
cls,
key: tuple[asyncio.AbstractEventLoop, str, str],
task: asyncio.Task[Optional[PluginIndex]],
) -> None:
"""异步索引请求结束后释放事件循环和仓库引用。"""
with cls._index_task_lock:
if cls._index_tasks.get(key) is task:
cls._index_tasks.pop(key, None)
# 调用方可能在请求完成前取消等待；主动读取异常避免事件循环告警，
# 不影响其他仍在等待同一个任务的调用方取得原始异常。
try:
task.exception()
except BaseException: # noqa: BLE001 - 仅消费已完成任务的异常
pass
@staticmethod
async def _acquire_index_fetch_slot() -> None:
"""以可取消方式等待跨同步/异步调用共享的索引请求闸门。"""
while not _PLUGIN_INDEX_FETCH_GATE.acquire(blocking=False):
await asyncio.sleep(0.01)
def _fetch_plugin_index_sync(
self,
package_url: str,
headers: Optional[dict[str, str]],
) -> Optional[PluginIndex]:
"""在共享进程级闸门内完成一次同步索引读取。"""
with _PLUGIN_INDEX_FETCH_GATE:
response = self.__request_plugin_index_with_fallback(
package_url,
headers=headers,
)
return self._resolve_plugin_index_result(response)
async def _fetch_plugin_index_async(
self,
package_url: str,
headers: Optional[dict[str, str]],
) -> Optional[PluginIndex]:
"""在共享进程级闸门内完成一次异步索引读取。"""
await self._acquire_index_fetch_slot()
try:
response = await self.__async_request_plugin_index_with_fallback(
package_url,
headers=headers,
)
finally:
_PLUGIN_INDEX_FETCH_GATE.release()
return self._resolve_plugin_index_result(response)
@cached(maxsize=1024, ttl=1800, skip_none=False) # type: ignore[misc]
def get_plugin_index_result(
self,
@@ -1148,11 +1216,28 @@ class PluginMarketTransport(metaclass=WeakSingleton):
if request is None:
raise ValueError("插件仓库地址无效")
package_url, headers = request
response = self.__request_plugin_index_with_fallback(
package_url,
headers=headers,
)
return self._resolve_plugin_index_result(response)
key = self._plugin_index_key(repo_url, package_version)
with self._index_future_lock:
future = self._index_futures.get(key)
owner = future is None
if owner:
future = Future()
self._index_futures[key] = future
if not owner:
return future.result()
try:
result = self._fetch_plugin_index_sync(package_url, headers)
except BaseException as error: # noqa: BLE001 - 共享结果必须传播原始异常
future.set_exception(error)
future.exception()
raise
else:
future.set_result(result)
return result
finally:
self._remove_index_future(key, future)
def get_plugins(self, repo_url: str,
package_version: Optional[str] = None) -> Optional[PluginIndex]:
@@ -1486,11 +1571,32 @@ class PluginMarketTransport(metaclass=WeakSingleton):
if request is None:
raise ValueError("插件仓库地址无效")
package_url, headers = request
response = await self.__async_request_plugin_index_with_fallback(
package_url,
headers=headers,
repo_key, generation_key = self._plugin_index_key(
repo_url,
package_version,
)
return self._resolve_plugin_index_result(response)
loop = asyncio.get_running_loop()
task_key = (loop, repo_key, generation_key)
with self._index_task_lock:
task = self._index_tasks.get(task_key)
if task is None:
task = cast(
asyncio.Task[Optional[PluginIndex]],
get_task_registry().create(
self._fetch_plugin_index_async(package_url, headers),
owner="plugin.market.index",
),
)
self._index_tasks[task_key] = task
task.add_done_callback(
lambda completed_task, key=task_key: self._remove_index_task(
key,
completed_task,
)
)
# 单个调用方取消等待时不能连带取消共享请求。
return await asyncio.shield(task)
async def async_get_plugins(self, repo_url: str,
package_version: Optional[str] = None) -> Optional[PluginIndex]:

33
app/domain/plugin.py
View File

@@ -19,6 +19,39 @@ PLUGIN_GENERATION_COMPATIBILITY: dict[str, tuple[str, ...]] = {
}
def normalize_plugin_market_repo_url(repo_url: str) -> str | None:
"""规范化 GitHub 插件市场地址，便于配置读取和请求去重。"""
value = str(repo_url or "").strip().rstrip("/")
if not value:
return None
value = value.removesuffix(".git")
try:
parsed_url = urlsplit(value)
except ValueError:
return None
if parsed_url.scheme not in {"http", "https"}:
return None
if (parsed_url.hostname or "").lower() != "github.com":
return None
parts = [item for item in parsed_url.path.split("/") if item]
if len(parts) < 2:
return None
return f"https://github.com/{parts[0]}/{parts[1]}"
def split_plugin_market_repo_urls(value: str | None) -> list[str]:
"""拆分插件市场配置并按大小写不敏感规则保持顺序去重。"""
repos: list[str] = []
seen_repos: set[str] = set()
for item in re.split(r"[\n,，]+", value or ""):
normalized_repo = normalize_plugin_market_repo_url(item)
if not normalized_repo or normalized_repo.lower() in seen_repos:
continue
repos.append(normalized_repo)
seen_repos.add(normalized_repo.lower())
return repos
@dataclass(frozen=True, slots=True)
class PluginReleaseInstallPlan:
"""描述远端插件内容准备模式，不持有同步或异步 I/O 实现。"""

9
app/runtime/extensions/plugin/catalog.py
View File

@@ -6,6 +6,7 @@ import importlib.util
from collections.abc import Callable, Mapping
from typing import Any, Optional
from app.domain.plugin import split_plugin_market_repo_urls
from app.foundation.version import compare_version
from app.runtime.extensions.plugin.contracts import supports_plugin_hook
from app.runtime.extensions.plugin.storage import PluginStorage
@@ -57,7 +58,7 @@ class PluginCatalogFacade:
plugin_market = get_runtime_setting('PLUGIN_MARKET')
if not plugin_market:
return []
markets = [item for item in plugin_market.split(",") if item]
markets = split_plugin_market_repo_urls(plugin_market)
result = self._market_catalog().collect(
markets=markets,
compatible_flags=self._system().compatible_flags(
@@ -234,7 +235,7 @@ class PluginCatalogFacade:
if progress_callback:
progress_callback(value=100, text="未配置插件市场，跳过刷新")
return []
markets = [item for item in plugin_market.split(",") if item]
markets = split_plugin_market_repo_urls(plugin_market)
result = await self._market_catalog().async_collect(
markets=markets,
compatible_flags=self._system().compatible_flags(
@@ -252,7 +253,7 @@ class PluginCatalogFacade:
plugin_market = get_runtime_setting('PLUGIN_MARKET')
if not plugin_market:
return []
markets = [item for item in plugin_market.split(",") if item]
markets = split_plugin_market_repo_urls(plugin_market)
result: list[Plugin] = await self._market_catalog().async_collect(
markets=markets,
compatible_flags=self._system().compatible_flags(
@@ -280,7 +281,7 @@ class PluginCatalogFacade:
def merge(self, higher: list[Plugin], base: list[Plugin]) -> list[Plugin]:
"""合并不同代际插件目录并保留市场优先级。"""
plugin_market = get_runtime_setting('PLUGIN_MARKET')
markets = [item for item in plugin_market.split(",") if item]
markets = split_plugin_market_repo_urls(plugin_market)
return self._market_catalog().merge(higher, base, markets)
def _safe_state(self, plugin_id: str, plugin: Any) -> bool:

2
docs/architecture-overview.md
View File

@@ -755,7 +755,7 @@ flowchart LR
| 指标 | 当前值 |
|---|---:|
| Python 模块 | 985 |
| 内部导入边 | 8,356 |
| 内部导入边 | 8,358 |
| 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
| Application / Chain 具体 Adapter 直连 | 0 / 0 |
| Direct egress | 53（债务已清零，53 条精确 containment） |

2
docs/architecture/optimization-checklist.md
View File

@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
| 指标 | 当前值 | 解释 |
|---|---:|---|
| 宿主 Python 模块 / 内部依赖边 | 985 / 8,356 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索与整理任务恢复模块的受控依赖 |
| 宿主 Python 模块 / 内部依赖边 | 985 / 8,358 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索与整理任务恢复模块的受控依赖 |
| 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
| 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
| Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

6
tests/fixtures/architecture/dependency-baseline.json vendored
View File

@@ -1074,8 +1074,8 @@
"runtime_only": true
}
},
"edge_count": 8356,
"edge_sha256": "62b05583fcf296d4f33f9c5ba77eeb8f5a8a200a3809e74ad4df28726587ece4",
"edge_count": 8358,
"edge_sha256": "c0785c61259ea746f28fa0e13d65fb72c3d0d438d52bc624177f0787f98f8728",
"edges": [
"app -> app.foundation",
"app -> app.foundation.environment",
@@ -7975,6 +7975,8 @@
"app.runtime.extensions.module.manager -> app.schemas.types",
"app.runtime.extensions.plugin.admission -> app.schemas",
"app.runtime.extensions.plugin.admission -> app.schemas.exception",
"app.runtime.extensions.plugin.catalog -> app.domain",
"app.runtime.extensions.plugin.catalog -> app.domain.plugin",
"app.runtime.extensions.plugin.catalog -> app.foundation",
"app.runtime.extensions.plugin.catalog -> app.foundation.version",
"app.runtime.extensions.plugin.catalog -> app.runtime",

48
tests/test_plugin_catalog_runtime.py
View File

@@ -2,11 +2,59 @@
from types import SimpleNamespace
import pytest
from app.runtime.extensions.plugin.catalog import PluginCatalogFacade
from app.schemas.plugin import PluginRuntimeStatus
from app.schemas.types import SystemConfigKey
@pytest.mark.asyncio
async def test_async_online_normalizes_market_configuration(monkeypatch) -> None:
"""目录入口与候选库存使用相同的规范化市场列表。"""
from app.runtime.extensions.plugin import catalog as catalog_module
requested_markets: list[str] = []
class FakeMarketCatalog:
async def async_collect(self, **kwargs):
requested_markets.extend(kwargs["markets"])
return []
monkeypatch.setattr(
catalog_module,
"get_runtime_setting",
lambda key: {
"PLUGIN_MARKET": (
" https://github.com/Example/Plugins.git/ ,"
"https://github.com/example/plugins"
),
"VERSION_FLAG": "v3",
}[key],
)
facade = PluginCatalogFacade(
classes=lambda: {},
running=lambda: {},
storage=lambda: SimpleNamespace(read=lambda _key: []),
system=lambda: SimpleNamespace(
compatible_flags=lambda _flag: [],
),
market_catalog=lambda: FakeMarketCatalog(),
market_loader=lambda *_args, **_kwargs: [],
async_market_loader=lambda *_args, **_kwargs: [],
map_plugin=lambda **_kwargs: None,
auth_checker=lambda **_kwargs: True,
plugin_attr=lambda _plugin_id, _attr: None,
plugin_instance=lambda _plugin_id: None,
plugin_instances=lambda: {},
runtime_status=lambda _plugin_id: None,
log=SimpleNamespace(info=lambda *_args: None),
)
assert await facade.async_online() == []
assert requested_markets == ["https://github.com/Example/Plugins"]
def test_installed_catalog_keeps_plugins_that_are_not_loaded():
"""已安装清单中的插件即使缺依赖或源码也必须保留可观察卡片。"""
class ActivePlugin:

100
tests/test_plugin_market_index_policy.py
View File

@@ -1,10 +1,13 @@
"""插件市场索引同步与异步策略一致性测试。"""
import asyncio
import json
import threading
from contextlib import asynccontextmanager, contextmanager
import pytest
from app.adapters.external.plugin import client as plugin_client_module
from app.adapters.external.plugin.client import (
PLUGIN_INDEX_MAX_BYTES,
PLUGIN_INDEX_MAX_ENTRIES,
@@ -13,6 +16,7 @@ from app.adapters.external.plugin.client import (
PluginMarketTransport,
_format_request_error,
)
from app.domain.plugin import split_plugin_market_repo_urls
SYNC_INDEX_REQUEST = "_PluginMarketTransport__request_plugin_index_with_fallback"
ASYNC_INDEX_REQUEST = "_PluginMarketTransport__async_request_plugin_index_with_fallback"
@@ -23,6 +27,18 @@ def test_plugin_index_request_timeout_is_bounded() -> None:
assert PLUGIN_INDEX_REQUEST_TIMEOUT == 15
def test_plugin_market_repo_config_is_normalized_once() -> None:
"""市场配置的空白、尾斜杠和重复仓库不能制造重复读取任务。"""
assert split_plugin_market_repo_urls(
" https://github.com/Example/Plugins.git/ ,"
"https://github.com/example/plugins,"
"https://github.com/other/plugins/ "
) == [
"https://github.com/Example/Plugins",
"https://github.com/other/plugins",
]
def test_github_request_strategies_do_not_retry_direct_when_proxy_is_configured(
monkeypatch,
) -> None:
@@ -676,6 +692,90 @@ async def test_async_plugin_index_result_preserves_absent_state(monkeypatch) ->
assert result is None
@pytest.mark.asyncio
async def test_async_plugin_index_requests_are_single_flight(monkeypatch) -> None:
"""同一仓库和代际的并发索引读取只能产生一次出站请求。"""
helper = PluginMarketTransport()
repo_url = "https://github.com/single-flight-owner/single-flight-repo"
started = asyncio.Event()
release = asyncio.Event()
requests = 0
async def request(_url: str, *, headers: dict):
nonlocal requests
requests += 1
started.set()
await release.wait()
return 200, '{"DemoPlugin": {"version": "1.2.3"}}'
monkeypatch.setattr(helper, ASYNC_INDEX_REQUEST, request)
await helper.async_get_plugin_index_result.cache_clear()
first = asyncio.create_task(
helper.async_get_plugin_index_result(repo_url, "v3")
)
await asyncio.wait_for(started.wait(), timeout=1)
second = asyncio.create_task(
helper.async_get_plugin_index_result(
"https://github.com/single-flight-owner/single-flight-repo/",
"v3",
)
)
await asyncio.sleep(0)
release.set()
assert await asyncio.gather(first, second) == [
{"DemoPlugin": {"version": "1.2.3"}},
{"DemoPlugin": {"version": "1.2.3"}},
]
assert requests == 1
@pytest.mark.asyncio
async def test_async_plugin_index_requests_share_process_gate(monkeypatch) -> None:
"""不同目录刷新入口也必须受同一个进程级索引请求闸门约束。"""
helper = PluginMarketTransport()
monkeypatch.setattr(
plugin_client_module,
"_PLUGIN_INDEX_FETCH_GATE",
threading.BoundedSemaphore(2),
)
active = 0
peak = 0
started = asyncio.Event()
release = asyncio.Event()
async def request(_url: str, *, headers: dict):
nonlocal active, peak
active += 1
peak = max(peak, active)
if active == 2:
started.set()
await release.wait()
active -= 1
return 200, '{"DemoPlugin": {"version": "1.2.3"}}'
monkeypatch.setattr(helper, ASYNC_INDEX_REQUEST, request)
await helper.async_get_plugin_index_result.cache_clear()
tasks = [
asyncio.create_task(
helper.async_get_plugin_index_result(
f"https://github.com/process-gate-owner/repository-{index}",
"v3",
)
)
for index in range(4)
]
await asyncio.wait_for(started.wait(), timeout=1)
assert active == 2
release.set()
results = await asyncio.gather(*tasks)
assert peak == 2
assert len(results) == 4
def test_plugin_index_result_propagates_adapter_exception(monkeypatch) -> None:
"""请求异常必须传播给应用库存统一转换为失败事实。"""
helper = PluginMarketTransport()