Merge pull request #6406 from jxxghp/fix/site-connectivity-explicit-transactions

Fix/site connectivity explicit transactions

app/db/oper/site.py

@@ -2,6 +2,7 @@ from datetime import datetime
 from typing import Any, List, Mapping, Tuple, Optional
 
 from sqlalchemy import delete as sqlalchemy_delete
+from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
 from app.db.models.site import Site
@@ -285,79 +286,90 @@ class SiteOper(DbOper):
         更新站点图标
         """
         icon_base64 = f"data:image/ico;base64,{icon_base64}" if icon_base64 else ""
-        siteicon = self.get_icon_by_domain(domain)
-        if not siteicon:
-            self._stage_create(
-                SiteIcon(name=name, domain=domain, url=icon_url, base64=icon_base64)
-            )
-        elif icon_base64:
-            self._stage_update(siteicon, {
-                "url": icon_url,
-                "base64": icon_base64
-            })
+
+        def write(db: Session) -> None:
+            """在同一同步事务中查询并更新站点图标。"""
+            siteicon = SiteIcon.get_by_domain(db, domain)
+            if not siteicon:
+                db.add(SiteIcon(
+                    name=name,
+                    domain=domain,
+                    url=icon_url,
+                    base64=icon_base64,
+                ))
+            elif icon_base64:
+                siteicon.url = icon_url
+                siteicon.base64 = icon_base64
+
+        self._execute_sync_write(write)
         return True
 
     def success(self, domain: str, seconds: Optional[int] = None):
         """
         站点访问成功
         """
-        lst_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-        sta = SiteStatistic.get_by_domain(self._db, domain)
-        if sta:
-            # 使用深复制确保 note 是全新的字典对象
-            note = dict(sta.note) if sta.note else {}
-            avg_seconds = None
+        def write(db: Session) -> None:
+            """在同一同步事务中读取并更新站点统计。"""
+            lst_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            sta = SiteStatistic.get_by_domain(db, domain)
+            if sta:
+                # 使用深复制确保 note 是全新的字典对象
+                note = dict(sta.note) if sta.note else {}
+                avg_seconds = None
 
-            if seconds is not None:
-                note[lst_date] = seconds or 1
-                avg_times = len(note.keys())
-                if avg_times > 10:
-                    note = dict(sorted(note.items(), key=lambda x: x[0], reverse=True)[:10])
-                avg_seconds = sum([v for v in note.values()]) // avg_times
+                if seconds is not None:
+                    note[lst_date] = seconds or 1
+                    avg_times = len(note.keys())
+                    if avg_times > 10:
+                        note = dict(sorted(note.items(), key=lambda x: x[0], reverse=True)[:10])
+                    avg_seconds = sum([v for v in note.values()]) // avg_times
 
-            self._stage_update(sta, {
-                "success": sta.success + 1,
-                "seconds": avg_seconds or sta.seconds,
-                "lst_state": 0,
-                "lst_mod_date": lst_date,
-                "note": note
-            })
-        else:
-            note = {}
-            if seconds is not None:
-                note = {
-                    lst_date: seconds or 1
-                }
-            self._stage_create(SiteStatistic(
-                domain=domain,
-                success=1,
-                fail=0,
-                seconds=seconds or 1,
-                lst_state=0,
-                lst_mod_date=lst_date,
-                note=note
-            ))
+                for key, value in {
+                    "success": sta.success + 1,
+                    "seconds": avg_seconds or sta.seconds,
+                    "lst_state": 0,
+                    "lst_mod_date": lst_date,
+                    "note": note,
+                }.items():
+                    setattr(sta, key, value)
+            else:
+                note = {}
+                if seconds is not None:
+                    note = {lst_date: seconds or 1}
+                db.add(SiteStatistic(
+                    domain=domain,
+                    success=1,
+                    fail=0,
+                    seconds=seconds or 1,
+                    lst_state=0,
+                    lst_mod_date=lst_date,
+                    note=note,
+                ))
+
+        self._execute_sync_write(write)
 
     def fail(self, domain: str):
         """
         站点访问失败
         """
-        lst_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-        sta = SiteStatistic.get_by_domain(self._db, domain)
-        if sta:
-            self._stage_update(sta, {
-                "fail": sta.fail + 1,
-                "lst_state": 1,
-                "lst_mod_date": lst_date
-            })
-        else:
-            self._stage_create(SiteStatistic(
-                domain=domain,
-                success=0,
-                fail=1,
-                lst_state=1,
-                lst_mod_date=lst_date
-            ))
+        def write(db: Session) -> None:
+            """在同一同步事务中读取并更新站点失败统计。"""
+            lst_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            sta = SiteStatistic.get_by_domain(db, domain)
+            if sta:
+                sta.fail += 1
+                sta.lst_state = 1
+                sta.lst_mod_date = lst_date
+            else:
+                db.add(SiteStatistic(
+                    domain=domain,
+                    success=0,
+                    fail=1,
+                    lst_state=1,
+                    lst_mod_date=lst_date,
+                ))
+
+        self._execute_sync_write(write)
 
     async def async_success(self, domain: str, seconds: Optional[int] = None):
         """

app/startup/modules_initializer.py

@@ -130,6 +130,7 @@ from app.startup.subscription import (
 )
 from app.startup.chain_events import TransactionalChainDurableEventWriter
 from app.startup.download_failure import TransactionalDownloadFailureRepository
+from app.startup.site import TransactionalSiteRepository
 from app.startup.workflow import TransactionalWorkflowExecutionService
 from app.startup.transaction import TransactionalWriteRunner
 from app.startup.context import (
@@ -685,7 +686,10 @@ async def init_modules() -> HostRuntime:
     workflow_execution = TransactionalWorkflowExecutionService(SessionFactory)
     configure_workflow_legacy_writer(workflow_execution)
     configure_chain_data_ports(
-        site=lambda: SiteOper(),
+        site=lambda: TransactionalSiteRepository(
+            sync_session=SessionFactory,
+            async_session=async_session_scope,
+        ),
         subscribe=lambda: SubscribeOper(),
         workflow=lambda: WorkflowOper(),
         download_history=lambda: DownloadHistoryOper(),
@@ -719,14 +723,23 @@ async def init_modules() -> HostRuntime:
     )
     configure_passkey_service(PasskeyService(repository=PassKeyOper()))
     configure_transfer_history_provider(lambda: TransferHistoryOper())
-    configure_site_query_service(SiteQueryService(repository=SiteOper()))
-    configure_site_health_service(SiteHealthService(repository=SiteOper()))
+    configure_site_query_service(SiteQueryService(repository=TransactionalSiteRepository(
+        sync_session=SessionFactory,
+        async_session=async_session_scope,
+    )))
+    configure_site_health_service(SiteHealthService(repository=TransactionalSiteRepository(
+        sync_session=SessionFactory,
+        async_session=async_session_scope,
+    )))
     configure_workflow_query(WorkflowQueryService(repository=WorkflowOper()))
     configure_agent_data_ports(
         agent_chat=lambda: AgentChatOper(),
         agent_task=lambda: AgentTaskOper(),
         user=lambda: UserOper(),
-        site=lambda: SiteOper(),
+        site=lambda: TransactionalSiteRepository(
+            sync_session=SessionFactory,
+            async_session=async_session_scope,
+        ),
         subscribe=lambda: SubscribeOper(),
         subscribe_history=lambda: SubscribeHistoryOper(),
         transfer_history=lambda: TransferHistoryOper(),

app/startup/site.py

@@ -0,0 +1,225 @@
+"""站点 Chain 端口的显式会话与事务适配器。"""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from contextlib import AbstractAsyncContextManager
+from typing import Any, TypeVar
+
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session
+
+from app.db.oper.site import SiteOper
+from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork
+
+
+T = TypeVar("T")
+
+
+class TransactionalSiteRepository:
+    """为同步 Chain 站点端口和异步健康统计提供短生命周期会话。"""
+
+    def __init__(
+        self,
+        *,
+        sync_session: Callable[[], Session],
+        async_session: Callable[[], AbstractAsyncContextManager[AsyncSession]],
+    ) -> None:
+        """保存同步会话工厂和异步会话上下文工厂。"""
+        self._sync_session = sync_session
+        self._async_session = async_session
+
+    def _read(self, operation: Callable[[SiteOper], T]) -> T:
+        """在独立同步会话中执行只读站点操作。"""
+        with self._sync_session() as session:
+            return operation(SiteOper(db=session))
+
+    def _write(self, operation: Callable[[SiteOper], T]) -> T:
+        """在独立同步 UoW 中执行站点写操作。"""
+        with self._sync_session() as session:
+            session.expire_on_commit = False
+            unit_of_work = SqlAlchemyUnitOfWork(session)
+            try:
+                result = operation(SiteOper(db=session))
+                unit_of_work.commit()
+                return result
+            except Exception:
+                unit_of_work.rollback()
+                raise
+
+    async def _async_write(
+        self,
+        operation: Callable[[SiteOper], Awaitable[T]],
+    ) -> T:
+        """在独立异步 UoW 中执行站点写操作。"""
+        async with self._async_session() as session:
+            session.sync_session.expire_on_commit = False
+            unit_of_work = SqlAlchemyAsyncUnitOfWork(session)
+            try:
+                result = await operation(SiteOper(db=session))
+                await unit_of_work.commit()
+                return result
+            except Exception:
+                await unit_of_work.rollback()
+                raise
+
+    async def _async_read(self, operation: Callable[[SiteOper], Awaitable[T]]) -> T:
+        """在独立异步会话中执行只读站点操作。"""
+        async with self._async_session() as session:
+            return await operation(SiteOper(db=session))
+
+    def add(self, **kwargs: Any) -> tuple[bool, str]:
+        """新增站点并提交事务。"""
+        return self._write(lambda repository: repository.add(**kwargs))
+
+    def get(self, site_id: int) -> Any:
+        """按 ID 查询站点。"""
+        return self._read(lambda repository: repository.get(site_id))
+
+    def get_by_domain(self, domain: str) -> Any:
+        """按域名查询站点。"""
+        return self._read(lambda repository: repository.get_by_domain(domain))
+
+    def get_domains_by_ids(self, ids: list[int]) -> list[str | None]:
+        """查询一组站点 ID 对应的域名。"""
+        return self._read(lambda repository: repository.get_domains_by_ids(ids))
+
+    def list(self) -> list[Any]:
+        """查询全部站点。"""
+        return self._read(lambda repository: repository.list())
+
+    def list_order_by_pri(self) -> list[Any]:
+        """同步按优先级查询站点。"""
+        return self._read(lambda repository: repository.list_order_by_pri())
+
+    def get_userdata_latest(self) -> list[Any]:
+        """同步查询各站点最新用户数据。"""
+        return self._read(lambda repository: repository.get_userdata_latest())
+
+    async def async_get(self, site_id: int) -> Any:
+        """异步按 ID 查询站点。"""
+        return await self._async_read(lambda repository: repository.async_get(site_id))
+
+    async def async_get_by_domain(self, domain: str) -> Any:
+        """异步按域名查询站点。"""
+        return await self._async_read(
+            lambda repository: repository.async_get_by_domain(domain)
+        )
+
+    async def async_get_by_name(self, name: str) -> Any:
+        """异步按名称查询站点。"""
+        return await self._async_read(
+            lambda repository: repository.async_get_by_name(name)
+        )
+
+    async def async_list(self) -> list[Any]:
+        """异步查询全部站点。"""
+        return await self._async_read(lambda repository: repository.async_list())
+
+    async def async_list_order_by_pri(self) -> list[Any]:
+        """异步按优先级查询站点。"""
+        return await self._async_read(
+            lambda repository: repository.async_list_order_by_pri()
+        )
+
+    async def async_update(self, site_id: int, payload: dict[str, Any]) -> Any:
+        """异步更新站点并提交事务。"""
+        return await self._async_write(
+            lambda repository: repository.async_update(site_id, payload)
+        )
+
+    async def async_get_userdata_by_domain(
+        self,
+        domain: str,
+        workdate: str | None = None,
+    ) -> list[Any]:
+        """异步查询站点用户数据。"""
+        return await self._async_read(
+            lambda repository: repository.async_get_userdata_by_domain(domain, workdate)
+        )
+
+    async def async_get_userdata_latest(self) -> list[Any]:
+        """异步查询各站点最新用户数据。"""
+        return await self._async_read(
+            lambda repository: repository.async_get_userdata_latest()
+        )
+
+    async def async_get_icon_by_domain(self, domain: str) -> Any:
+        """异步按域名查询站点图标。"""
+        return await self._async_read(
+            lambda repository: repository.async_get_icon_by_domain(domain)
+        )
+
+    async def async_get_statistic_by_domain(self, domain: str) -> Any:
+        """异步按域名查询站点统计。"""
+        return await self._async_read(
+            lambda repository: repository.async_get_statistic_by_domain(domain)
+        )
+
+    async def async_list_statistics(self) -> list[Any]:
+        """异步查询全部站点统计。"""
+        return await self._async_read(
+            lambda repository: repository.async_list_statistics()
+        )
+
+    def update(self, site_id: int, payload: dict[str, Any]) -> Any:
+        """更新站点并提交事务。"""
+        return self._write(lambda repository: repository.update(site_id, payload))
+
+    def update_cookie(self, domain: str, cookies: str) -> tuple[bool, str]:
+        """更新站点 Cookie 并提交事务。"""
+        return self._write(
+            lambda repository: repository.update_cookie(domain, cookies)
+        )
+
+    def update_rss(self, domain: str, rss: str) -> tuple[bool, str]:
+        """更新站点 RSS 地址并提交事务。"""
+        return self._write(lambda repository: repository.update_rss(domain, rss))
+
+    def update_userdata(
+        self,
+        domain: str,
+        name: str,
+        payload: dict[str, Any],
+    ) -> tuple[bool, str]:
+        """更新站点用户数据并提交事务。"""
+        return self._write(
+            lambda repository: repository.update_userdata(domain, name, payload)
+        )
+
+    def update_icon(
+        self,
+        name: str,
+        domain: str,
+        icon_url: str,
+        icon_base64: str,
+    ) -> bool:
+        """更新站点图标并提交事务。"""
+        return self._write(
+            lambda repository: repository.update_icon(
+                name,
+                domain,
+                icon_url,
+                icon_base64,
+            )
+        )
+
+    def success(self, domain: str, seconds: int | None = None) -> Any:
+        """记录站点访问成功并提交事务。"""
+        return self._write(lambda repository: repository.success(domain, seconds))
+
+    def fail(self, domain: str) -> Any:
+        """记录站点访问失败并提交事务。"""
+        return self._write(lambda repository: repository.fail(domain))
+
+    async def async_success(self, domain: str, seconds: int | None = None) -> Any:
+        """异步记录站点访问成功并提交事务。"""
+        return await self._async_write(
+            lambda repository: repository.async_success(domain, seconds)
+        )
+
+    async def async_fail(self, domain: str) -> Any:
+        """异步记录站点访问失败并提交事务。"""
+        return await self._async_write(
+            lambda repository: repository.async_fail(domain)
+        )

tests/conftest.py

@@ -148,6 +148,8 @@ def configure_plugin_system_services():
     from app.db.oper.message import MessageOper
     from app.db.oper.passkey import PassKeyOper
     from app.startup.subscription import TransactionalSubscribeWriter
+    from app.startup.download_failure import TransactionalDownloadFailureRepository
+    from app.startup.site import TransactionalSiteRepository
     from app.startup.workflow import TransactionalWorkflowExecutionService
     from app.startup.transaction import TransactionalWriteRunner
 
@@ -202,15 +204,24 @@ def configure_plugin_system_services():
         )
     )
 
+    def site_repository() -> TransactionalSiteRepository:
+        """按生产组合根方式创建显式事务站点仓储。"""
+        return TransactionalSiteRepository(
+            sync_session=SessionFactory,
+            async_session=async_session_scope,
+        )
+
     configure_chain_data_ports(
-        site=lambda: SiteOper(),
+        site=site_repository,
         subscribe=lambda: SubscribeOper(),
         workflow=lambda: WorkflowOper(),
         download_history=lambda: DownloadHistoryOper(),
         transfer_history=lambda: TransferHistoryOper(),
         transfer_pending=lambda: TransferPendingOper(),
         media_server=lambda: MediaServerOper(),
-        download_failure=lambda: DownloadFailureOper(),
+        download_failure=lambda: TransactionalDownloadFailureRepository(
+            SessionFactory
+        ),
         user=lambda: UserOper(),
     )
     configure_chain_runtime_context_provider(lambda: ChainRuntimeContext(
@@ -227,8 +238,8 @@ def configure_plugin_system_services():
         module_dispatcher_factory=ModuleInvocationDispatcher,
         configuration=build_chain_runtime_config(settings),
     ))
-    configure_site_query_service(SiteQueryService(repository=SiteOper()))
-    configure_site_health_service(SiteHealthService(repository=SiteOper()))
+    configure_site_query_service(SiteQueryService(repository=site_repository()))
+    configure_site_health_service(SiteHealthService(repository=site_repository()))
     configure_workflow_query(WorkflowQueryService(repository=WorkflowOper()))
     from app.db.oper.agenttask import AgentTaskOper
     from app.db.oper.plugindata import PluginDataOper
@@ -236,7 +247,7 @@ def configure_plugin_system_services():
         agent_chat=lambda: AgentChatOper(),
         agent_task=lambda: AgentTaskOper(),
         user=lambda: UserOper(),
-        site=lambda: SiteOper(),
+        site=site_repository,
         subscribe=lambda: SubscribeOper(),
         subscribe_history=lambda: SubscribeHistoryOper(),
         transfer_history=lambda: TransferHistoryOper(),

tests/fixtures/architecture/dependency-baseline.json

@@ -13,8 +13,8 @@
     "runtime_to_db": [],
     "workflow_to_db": []
   },
-  "edge_count": 6429,
-  "edge_sha256": "664681a500ba1c3d273568829b1b860e4dae7fe7a6c053c566583341fee7f903",
+  "edge_count": 6434,
+  "edge_sha256": "5c47cae41f5d90e1030757d0cb47db8eed373424dda5db17c900271e1d03a9c8",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -6168,6 +6168,7 @@
     "app.startup.modules_initializer -> app.startup.download_failure",
     "app.startup.modules_initializer -> app.startup.managed_resources_initializer",
     "app.startup.modules_initializer -> app.startup.outbox",
+    "app.startup.modules_initializer -> app.startup.site",
     "app.startup.modules_initializer -> app.startup.subscription",
     "app.startup.modules_initializer -> app.startup.transaction",
     "app.startup.modules_initializer -> app.startup.workflow",
@@ -6228,6 +6229,10 @@
     "app.startup.scheduler_initializer -> app.application",
     "app.startup.scheduler_initializer -> app.application.scheduling",
     "app.startup.scheduler_initializer -> app.scheduler",
+    "app.startup.site -> app.db",
+    "app.startup.site -> app.db.oper",
+    "app.startup.site -> app.db.oper.site",
+    "app.startup.site -> app.db.uow",
     "app.startup.subscription -> app.adapters",
     "app.startup.subscription -> app.adapters.external",
     "app.startup.subscription -> app.adapters.external.server",
@@ -6446,7 +6451,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 797,
+  "module_count": 798,
   "modules": [
     "app",
     "app.adapters",
@@ -7219,6 +7224,7 @@
     "app.startup.plugins_initializer",
     "app.startup.routers_initializer",
     "app.startup.scheduler_initializer",
+    "app.startup.site",
     "app.startup.subscription",
     "app.startup.transaction",
     "app.startup.transfer_initializer",

tests/test_db_oper_layer.py

@@ -204,6 +204,22 @@ def test_site_oper_update_icon_creates_then_only_overwrites_with_content(db):
     assert first.startswith("data:image/ico;base64,")
 
 
+def test_site_oper_icon_without_explicit_session_uses_transaction_runner(db):
+    """无显式会话的图标写入应由兼容事务执行器完成提交。"""
+    oper = SiteOper()
+
+    oper.update_icon(
+        "兼容站点",
+        "op-icon-no-session.test",
+        "https://op-icon-no-session.test/favicon.ico",
+        "AAA",
+    )
+
+    icon = SiteIcon.get_by_domain(db.session, "op-icon-no-session.test")
+    assert icon.name == "兼容站点"
+    assert icon.base64 == "data:image/ico;base64,AAA"
+
+
 def test_site_oper_success_accumulates_and_records_state(db):
     """
     访问成功累加计数并把最后状态标记为成功。
@@ -219,6 +235,17 @@ def test_site_oper_success_accumulates_and_records_state(db):
     assert stat.seconds
 
 
+def test_site_oper_statistics_without_explicit_session_use_transaction_runner(db):
+    """无显式会话的站点统计必须由兼容事务执行器完成提交。"""
+    oper = SiteOper()
+
+    oper.success("op-stat-no-session.test", seconds=3)
+    oper.fail("op-stat-no-session.test")
+
+    stat = SiteStatistic.get_by_domain(db.session, "op-stat-no-session.test")
+    assert (stat.success, stat.fail, stat.lst_state) == (1, 1, 1)
+
+
 def test_site_oper_success_caps_the_timing_note_at_ten_entries(db):
     """
     耗时记录最多保留最近 10 条，超出时丢弃最旧的。

tests/test_site_connectivity.py

@@ -0,0 +1,28 @@
+"""站点连通性测试链路的数据库回归测试。"""
+
+from app.chain.site import SiteChain
+from app.db.models.site import Site
+from app.db.models.sitestatistic import SiteStatistic
+
+
+def test_site_connectivity_records_result_without_injected_session(db, monkeypatch):
+    """默认站点端口完成测试后应提交统计，而不是对空会话调用 execute。"""
+    db.watermark(Site, SiteStatistic)
+    db.add(Site(
+        name="连通性测试站点",
+        domain="connectivity.test",
+        url="https://connectivity.test/",
+        is_active=True,
+    ))
+    monkeypatch.setattr(
+        SiteChain,
+        "_SiteChain__test",
+        lambda _self, _site: (True, "连接成功"),
+    )
+
+    status, message = SiteChain().test("https://connectivity.test/")
+
+    statistic = SiteStatistic.get_by_domain(db.session, "connectivity.test")
+    assert (status, message) == (True, "连接成功")
+    assert statistic.success == 1
+    assert statistic.lst_state == 0