fix(executor): task refs + pipeline semaphore + off-loop broker [修复组③]

backend/config.py

@@ -24,6 +24,15 @@ class Settings(BaseSettings):
     # Task execution mode: "local" (in-process asyncio) or "celery" (distributed)
     task_executor: Literal["local", "celery"] = "local"
 
+    # AUDIT C6: process-wide cap on concurrently-RUNNING pipeline executions in
+    # the local (in-process asyncio) executor — independent of the per-domain
+    # cap in pipeline/domain_limiter.py. Bounds how many schedules firing on
+    # the same tick plus manual/webhook triggers can drive Chrome/opencli
+    # subprocesses at once on the one event loop. Only meaningful when
+    # task_executor="local" (celery fans out across worker processes
+    # instead). Env: LOCAL_MAX_CONCURRENT_PIPELINES.
+    local_max_concurrent_pipelines: int = 8
+
     # Collection orchestrator:
     # admin — API内置 scheduler.py / Celery Beat 驱动定时采集（默认）
     # iii   — III engine + schedule-bootstrap 驱动 cron；API 仅保留 UI/手动任务

backend/executor/celery_exec.py

@@ -1,5 +1,6 @@
 """Celery-based distributed executor."""
 
+import asyncio
 import logging
 
 from backend.executor.base import AbstractExecutor
@@ -29,15 +30,23 @@ async def cancel_acquisition(self, execution_id: str) -> None:
 
     async def dispatch_collection(self, task_id: str, parameters: dict) -> dict:
         from backend.worker.tasks import run_collection
-        result = run_collection.apply_async(
-            kwargs={"task_id": task_id, "parameters": parameters}
+
+        # AUDIT C9: apply_async is a synchronous broker round-trip (network
+        # I/O to Redis/RabbitMQ). Called directly here it would block this
+        # event loop for every dispatch; to_thread moves the round-trip off
+        # the loop.
+        result = await asyncio.to_thread(
+            run_collection.apply_async,
+            kwargs={"task_id": task_id, "parameters": parameters},
         )
         return {"task_id": task_id, "celery_task_id": result.id}
 
     async def dispatch_scheduled_collection(
         self, schedule_id: str, source_id: str, parameters: dict
     ) -> None:
         from backend.worker.tasks import run_scheduled_collection
-        run_scheduled_collection.apply_async(
-            kwargs={"schedule_id": schedule_id, "source_id": source_id, "parameters": parameters}
+
+        await asyncio.to_thread(
+            run_scheduled_collection.apply_async,
+            kwargs={"schedule_id": schedule_id, "source_id": source_id, "parameters": parameters},
         )

backend/executor/local.py

@@ -3,10 +3,29 @@
 import asyncio
 import logging
 
+from backend.config import get_settings
 from backend.executor.base import AbstractExecutor
 
 logger = logging.getLogger(__name__)
 
+# AUDIT C6: process-wide cap on concurrently-RUNNING pipeline executions
+# (manual/webhook dispatch_collection + every dispatch_scheduled_collection),
+# independent of the per-domain cap in pipeline/domain_limiter.py. Keyed by
+# event loop id, mirroring domain_limiter._semaphore, so a semaphore is never
+# reused across event loops (production runs one loop and shares correctly;
+# tests get a fresh loop each and never touch a stale entry).
+_pipeline_semaphores: dict[int, asyncio.Semaphore] = {}
+
+
+def _pipeline_semaphore() -> asyncio.Semaphore:
+    loop = asyncio.get_running_loop()
+    key = id(loop)
+    sem = _pipeline_semaphores.get(key)
+    if sem is None:
+        sem = asyncio.Semaphore(get_settings().local_max_concurrent_pipelines)
+        _pipeline_semaphores[key] = sem
+    return sem
+
 
 def _log_task_exception(task: asyncio.Task) -> None:
     """Log any unhandled exception from a background asyncio task."""
@@ -21,6 +40,12 @@ def _log_task_exception(task: asyncio.Task) -> None:
 class LocalExecutor(AbstractExecutor):
     def __init__(self) -> None:
         self._acquisition_tasks: dict[str, asyncio.Task[None]] = {}
+        # AUDIT C5: distinct dicts from _acquisition_tasks — task_id (manual/
+        # webhook dispatch) and schedule_id (scheduled dispatch) are different
+        # id spaces, so each gets its own strong-reference table instead of
+        # sharing one dict with two key schemes.
+        self._collection_tasks: dict[str, asyncio.Task[dict]] = {}
+        self._scheduled_collection_tasks: dict[str, asyncio.Task[dict]] = {}
 
     async def dispatch_acquisition(self, execution_id: str) -> None:
         from backend.acquisition.runner import run_acquisition_execution
@@ -47,15 +72,59 @@ async def cancel_acquisition(self, execution_id: str) -> None:
         except asyncio.CancelledError:
             pass
 
-    async def dispatch_collection(self, task_id: str, parameters: dict) -> dict:
+    async def _run_collection(self, task_id: str, parameters: dict) -> dict:
         from backend.pipeline.runner import run_collection_pipeline
-        t = asyncio.create_task(run_collection_pipeline(task_id, parameters))
-        t.add_done_callback(_log_task_exception)
+
+        # AUDIT C6: the semaphore wraps execution, not dispatch — the task
+        # blocks on it here, inside itself, so dispatch_collection stays fast
+        # and non-blocking.
+        async with _pipeline_semaphore():
+            return await run_collection_pipeline(task_id, parameters)
+
+    async def _run_scheduled(
+        self, schedule_id: str, source_id: str, parameters: dict
+    ) -> dict:
+        from backend.pipeline.runner import run_scheduled_pipeline
+
+        async with _pipeline_semaphore():
+            return await run_scheduled_pipeline(schedule_id, source_id, parameters)
+
+    async def dispatch_collection(self, task_id: str, parameters: dict) -> dict:
+        # AUDIT C5: hold a strong reference exactly like dispatch_acquisition
+        # does. task_id is unique per call (each manual/webhook trigger has
+        # its own CollectionTask row), so unlike dispatch_scheduled_collection
+        # below there is no skip-if-inflight check — just track-and-pop. The
+        # task is stored before it can even attempt to acquire the semaphore,
+        # so it stays referenced (and un-GC'd) for the whole time it's
+        # blocked waiting on _pipeline_semaphore(), not just while running.
+        task = asyncio.create_task(self._run_collection(task_id, parameters))
+        task.add_done_callback(_log_task_exception)
+        self._collection_tasks[task_id] = task
+        task.add_done_callback(
+            lambda completed: self._collection_tasks.pop(task_id, None)
+            if self._collection_tasks.get(task_id) is completed
+            else None
+        )
         return {"task_id": task_id}
 
     async def dispatch_scheduled_collection(
         self, schedule_id: str, source_id: str, parameters: dict
     ) -> None:
-        from backend.pipeline.runner import run_scheduled_pipeline
-        t = asyncio.create_task(run_scheduled_pipeline(schedule_id, source_id, parameters))
-        t.add_done_callback(_log_task_exception)
+        # AUDIT C5: schedule_id has no per-run unique id at dispatch time (a
+        # schedule fires the same schedule_id on every tick), so this mirrors
+        # dispatch_acquisition's skip-if-inflight precedent — a schedule whose
+        # previous run hasn't finished yet is skipped rather than allowed to
+        # overlap itself.
+        current = self._scheduled_collection_tasks.get(schedule_id)
+        if current is not None and not current.done():
+            return
+        task = asyncio.create_task(
+            self._run_scheduled(schedule_id, source_id, parameters)
+        )
+        task.add_done_callback(_log_task_exception)
+        self._scheduled_collection_tasks[schedule_id] = task
+        task.add_done_callback(
+            lambda completed: self._scheduled_collection_tasks.pop(schedule_id, None)
+            if self._scheduled_collection_tasks.get(schedule_id) is completed
+            else None
+        )