From 1acf960312ef838fa1a10ad9aaf20abf406fd3c7 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 12:03:35 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=9A=E8=BF=9B=E7=A8=8B=E7=AB=AF=E5=88=B0?= =?UTF-8?q?=E7=AB=AF=E9=AA=8C=E6=94=B6=EF=BC=9Ascripts/test-multiprocess.s?= =?UTF-8?q?h=20=E7=94=A8=E6=9C=AC=E6=9C=BA=20postgres:16=20=E8=B5=B7?= =?UTF-8?q?=E4=B8=B4=E6=97=B6=E5=BA=93=EF=BC=8Ctests/test=5Fmultiprocess?= =?UTF-8?q?=5Fe2e.py=EF=BC=88=E7=BC=BA=E7=9C=81=E8=B7=B3=E8=BF=87=EF=BC=8C?= =?UTF-8?q?AGENTLAB=5FMULTIPROCESS=5FE2E=3D1=EF=BC=89=E5=9C=A8=E4=B8=8A?= =?UTF-8?q?=E9=9D=A2=E8=B5=B7=E4=B8=A4=E4=B8=AA=20uvicorn=20=E5=B7=A5?= =?UTF-8?q?=E4=BD=9C=E8=BF=9B=E7=A8=8B=EF=BC=8C=E7=94=A8=E6=BC=94=E7=A4=BA?= =?UTF-8?q?=E6=A8=A1=E5=9E=8B=E8=B7=91=E4=B8=83=E4=B8=AA=E5=9C=BA=E6=99=AF?= =?UTF-8?q?=E2=80=94=E2=80=94=E4=B8=A4=E4=B8=AA=E8=BF=9B=E7=A8=8B=E9=83=BD?= =?UTF-8?q?=E7=99=BB=E8=AE=B0=E3=80=81=E4=B8=80=E4=B8=AA=E5=BD=93=E7=8F=AD?= =?UTF-8?q?=EF=BC=9B=E8=BF=90=E8=A1=8C=E7=94=B1=E6=8E=A5=E5=88=B0=E8=AF=B7?= =?UTF-8?q?=E6=B1=82=E7=9A=84=E8=BF=9B=E7=A8=8B=E8=AE=A4=E9=A2=86=EF=BC=8C?= =?UTF-8?q?=E5=8F=A6=E4=B8=80=E4=B8=AA=E8=BF=9B=E7=A8=8B=E4=B8=8A=E7=9A=84?= =?UTF-8?q?=E6=B5=81=E6=94=B6=E5=BE=97=E5=88=B0=E5=AE=9E=E6=97=B6=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E5=92=8C=E6=B5=81=E5=BC=8F=E5=A2=9E=E9=87=8F=EF=BC=8C?= =?UTF-8?q?=E5=90=8C=E4=B8=80=E4=B8=AA=E5=AE=A1=E6=89=B9=E5=86=B3=E5=AE=9A?= =?UTF-8?q?=E5=90=8C=E6=97=B6=E5=8F=91=E7=BB=99=E4=B8=A4=E8=BE=B9=E5=8F=AA?= =?UTF-8?q?=E6=89=A7=E8=A1=8C=E4=B8=80=E6=AC=A1=EF=BC=8C=E4=B8=A4=E8=BE=B9?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E6=B5=81=E4=B8=8D=E9=87=8D=E4=B8=8D=E6=BC=8F?= =?UTF-8?q?=E3=80=81=E5=B0=81=E5=AD=98=E6=A0=B8=E5=AF=B9=E5=BE=97=E4=B8=8A?= =?UTF-8?q?=EF=BC=9B=E8=B7=A8=E8=BF=9B=E7=A8=8B=E5=81=9C=E6=AD=A2=EF=BC=9B?= =?UTF-8?q?=E5=AE=9A=E6=97=B6=E5=8F=AA=E8=A7=A6=E5=8F=91=E4=B8=80=E6=AC=A1?= =?UTF-8?q?=EF=BC=9B=E5=85=A8=E7=AB=99=E4=B8=80=E4=B8=AA=E5=90=8D=E9=A2=9D?= =?UTF-8?q?=E6=97=B6=E8=AF=84=E6=B5=8B=E7=9A=84=E4=BD=9C=E7=AD=94=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E6=8E=92=E7=9D=80=E9=98=9F=EF=BC=8C=E5=90=8E=E6=9D=A5?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E6=95=B0=E6=8D=AE=E8=BF=90=E8=A1=8C=E5=85=88?= =?UTF-8?q?=E6=8B=BF=E5=88=B0=E5=90=8D=E9=A2=9D=EF=BC=9B=E4=B8=A4=E4=B8=AA?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E5=90=8C=E6=97=B6=E8=AE=B0=E5=90=8C=E4=B8=80?= =?UTF-8?q?=E6=9C=AC=E6=AF=8F=E6=97=A5=E8=B4=A6=E4=B8=8D=E8=B6=8A=E8=BF=87?= =?UTF-8?q?=E4=B8=8A=E9=99=90=EF=BC=9B=E5=BC=BA=E6=9D=80=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E5=90=8E=E5=9C=A8=E8=B7=91=E7=9A=84=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E8=A2=AB=E7=85=A7=E5=AE=9E=E6=A0=87=E4=B8=AD=E6=96=AD?= =?UTF-8?q?=E3=80=81=E5=81=9C=E5=9C=A8=E5=AE=A1=E6=89=B9=E4=B8=8A=E7=9A=84?= =?UTF-8?q?=E7=85=A7=E5=B8=B8=E5=9C=A8=E5=8F=A6=E4=B8=80=E4=B8=AA=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E4=B8=8A=E6=89=B9=E5=AE=8C=E3=80=81=E5=BD=93=E7=8F=AD?= =?UTF-8?q?=E4=BA=A4=E6=8E=A5=E3=80=82=E9=A1=BA=E5=B8=A6=E4=BF=AE=EF=BC=9A?= =?UTF-8?q?=E7=AD=89=E5=90=8D=E9=A2=9D=E6=97=B6=E8=A2=AB=E5=81=9C=E6=AD=A2?= =?UTF-8?q?=E4=B9=9F=E7=85=A7=E5=AE=9E=E6=94=B6=E6=88=90=E5=B7=B2=E5=8F=96?= =?UTF-8?q?=E6=B6=88=EF=BC=88=E4=BB=A5=E5=89=8D=E5=81=9C=E5=9C=A8=E6=8E=92?= =?UTF-8?q?=E9=98=9F=E4=B8=AD=EF=BC=89=EF=BC=9B=E5=BD=93=E7=8F=AD=E7=9A=84?= =?UTF-8?q?=E6=9C=AC=E5=9C=B0=E4=BD=99=E9=87=8F=E4=B8=8D=E8=B6=85=E8=BF=87?= =?UTF-8?q?=E7=A7=9F=E7=BA=A6=E7=9A=84=E5=9B=9B=E5=88=86=E4=B9=8B=E4=B8=80?= =?UTF-8?q?=EF=BC=88=E7=A7=9F=E7=BA=A6=E8=B0=83=E7=9F=AD=E6=97=B6=E4=B8=80?= =?UTF-8?q?=E7=9B=B4=E4=B8=8D=E5=BD=93=E7=8F=AD=EF=BC=89=EF=BC=9B=E5=90=AF?= =?UTF-8?q?=E5=8A=A8=E6=B8=85=E7=90=86=E4=B8=8A=E4=BC=A0=E5=8D=8A=E6=88=90?= =?UTF-8?q?=E5=93=81=E6=97=B6=E6=B2=A1=E6=9C=89=E5=88=AB=E7=9A=84=E6=B4=BB?= =?UTF-8?q?=E7=9D=80=E7=9A=84=E8=BF=9B=E7=A8=8B=E5=B0=B1=E5=85=A8=E6=B8=85?= =?UTF-8?q?=EF=BC=9B=E6=96=87=E4=BB=B6=E9=94=81=E5=AE=88=E5=8D=AB=E7=9A=84?= =?UTF-8?q?=E5=87=A0=E9=A1=B9=E6=B5=8B=E8=AF=95=E5=8F=AA=E5=9C=A8=20SQLite?= =?UTF-8?q?=20=E8=B7=91=EF=BC=9B=E9=97=AE=E6=95=B0=E6=8D=AE=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E6=97=A5=E5=BF=97=E6=B8=85=E7=90=86=E9=82=A3=E4=B8=80?= =?UTF-8?q?=E6=8B=8D=E4=B8=8D=E5=B8=A6=E6=97=B6=E5=88=BB=E6=97=B6=E6=8C=89?= =?UTF-8?q?=E7=8E=B0=E5=9C=A8=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/chat/log.py | 7 +- backend/app/cluster/leader.py | 4 +- backend/app/cluster/service.py | 2 +- backend/app/data/table_versions.py | 15 +- backend/app/engine/runner.py | 27 +- backend/tests/test_multiprocess_e2e.py | 480 +++++++++++++++++++++ backend/tests/test_p3_api_errors.py | 4 + backend/tests/test_recipe_acceptance_p3.py | 1 + backend/tests/test_table_versions_p3.py | 2 + scripts/test-multiprocess.sh | 51 +++ 10 files changed, 585 insertions(+), 8 deletions(-) create mode 100644 backend/tests/test_multiprocess_e2e.py create mode 100755 scripts/test-multiprocess.sh diff --git a/backend/app/chat/log.py b/backend/app/chat/log.py index b6366414..c0e8a37c 100644 --- a/backend/app/chat/log.py +++ b/backend/app/chat/log.py @@ -242,13 +242,16 @@ async def purge(now: datetime, *, retain_days: float = RETAIN_DAYS, limit: int = _last_purge = 0.0 -async def purge_tick(now: datetime) -> list[str]: +async def purge_tick(now: datetime | None) -> list[str]: """调度循环每一拍调（scheduler/loop.process_due）：一小时最多清一次。""" global _last_purge if time.monotonic() - _last_purge < PURGE_EVERY_S and _last_purge: return [] _last_purge = time.monotonic() - return await purge(now) + # 调度循环那一拍不带时刻（process_due() 的 now 缺省为 None）：按现在算 + from app.db.base import utcnow + + return await purge(now or utcnow()) __all__ = ["FINISHED", "HEARTBEAT_S", "PING", "READ_BATCH", "RETAIN_DAYS", "STREAM_END", "add", "append", "beat", "forget", diff --git a/backend/app/cluster/leader.py b/backend/app/cluster/leader.py index 1662bfdf..ab2d800b 100644 --- a/backend/app/cluster/leader.py +++ b/backend/app/cluster/leader.py @@ -36,7 +36,9 @@ class Leadership: """本地看：还在当班（租约离到期还有余量）。""" if self.until is None: return False - return (now or utcnow()) + timedelta(seconds=MARGIN_S) < self.until + # 余量不超过租约的四分之一：租约调得很短（验收脚本）时也留得出当班的时间 + margin = min(MARGIN_S, lease_s() / 4) + return (now or utcnow()) + timedelta(seconds=margin) < self.until async def claim(self, now: datetime | None = None) -> bool: """接班或续租。交回这一拍之后是不是当班。""" diff --git a/backend/app/cluster/service.py b/backend/app/cluster/service.py index d4d5b39f..424996c8 100644 --- a/backend/app/cluster/service.py +++ b/backend/app/cluster/service.py @@ -80,7 +80,7 @@ class _Service: await caches.start() await self._lead() self.task = asyncio.create_task(self._loop(), name="agentlab-cluster-heartbeat") - logger.info("工作进程 %s 已登记（%s，租约 %.0f 秒、心跳 %.0f 秒；现在另有 %d 个进程）", identity.describe(identity.owner()), + logger.info("工作进程 %s 已登记（%s，租约 %g 秒、心跳 %g 秒；现在另有 %d 个进程）", identity.describe(identity.owner()), "PostgreSQL，可以多进程" if settings.db_dialect == "postgresql" else "SQLite，单进程", workers.lease_s(), workers.heartbeat_s(), len(live.others)) diff --git a/backend/app/data/table_versions.py b/backend/app/data/table_versions.py index e31d8f37..8f2f2fe4 100644 --- a/backend/app/data/table_versions.py +++ b/backend/app/data/table_versions.py @@ -2452,6 +2452,16 @@ def _same_path(a: str | os.PathLike[str], b: str | os.PathLike[str]) -> bool: TMP_STALE_S = 6 * 3600.0 +async def _tmp_min_age() -> float: + """启动清理半成品时只清多久以前的：SQLite（守卫保证只有本进程）、或者 PostgreSQL 上现在没有别的活着的进程，一个都不会有人 + 在写，全清；有别的进程活着，只清 TMP_STALE_S 以前的。""" + if settings.db_dialect != "postgresql": + return 0.0 + from app.cluster import state + + return TMP_STALE_S if (await state.current(fresh=True)).others else 0.0 + + def _sweep_tmp(min_age_s: float = 0.0) -> int: """删掉上次崩溃留下的 *.tmp-* 半成品（库和原件都是先写临时文件再 link）。min_age_s：只删改动时间早于这么多秒以前的 （多进程时别的进程可能正在写，见 startup）。""" @@ -2691,9 +2701,8 @@ async def startup(session: AsyncSession) -> None: logger.error("上传表格的版本存储被另一个进程占用，本进程跳过启动清理、迁移和回收") return async with _store_lock(): - # 半成品：单进程（SQLite，守卫保证只有本进程）启动时一个都不会有人在写，全清；PostgreSQL 可以多进程，别的进程正在解析的 - # 临时文件写在锁外，只清 TMP_STALE_S 以前的 - swept = await asyncio.to_thread(_sweep_tmp, TMP_STALE_S if settings.db_dialect == "postgresql" else 0.0) + # 半成品：没有别的进程在跑时一个都不会有人在写，全清；多进程时别的进程正在解析的临时文件写在锁外，只清旧的（_tmp_min_age） + swept = await asyncio.to_thread(_sweep_tmp, await _tmp_min_age()) if swept: logger.info("清理了 %d 个上次未完成的上传临时文件", swept) migrated = await _migrate_legacy(session) diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 73280840..43eff374 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -791,7 +791,14 @@ class RunManager: # 等待从 run.interrupted 算起——以前执行一直计到 finally，收尾那一截执行和等待两边都算， # active + wait 比墙钟还长（机器一忙就差出几十毫秒） paused: float | None = None - async with slots.hold(run_id, origin_kind): + # 先要并发名额（app/cluster/slots.py）。等名额的时候被停止（或者关停）也要照实收尾，见 _stopped_in_queue + slot = slots.hold(run_id, origin_kind) + try: + await slot.__aenter__() + except asyncio.CancelledError: + await self._stopped_in_queue(run_id) + raise + try: # 执行时长从拿到名额算起：排队等名额的那一段不算执行（墙钟从 started_at 算，started_at 也是拿到名额之后才写） started = time.perf_counter() try: @@ -976,6 +983,24 @@ class RunManager: from app.sandbox.manager import sandbox_manager await sandbox_manager.cleanup(run_id) + finally: + await slot.__aexit__(None, None, None) + + async def _stopped_in_queue(self, run_id: str) -> None: + """还在等并发名额时被取消。关停：什么都不写（还是排队中 / 刚认领的样子），照「属主不在了」的规矩由下次启动或别的 + 进程接手（排队的标失败、认领了的标中断）；人点了停止：收成 cancelled，同执行中停止（run.cancelled 带上是谁停的、封存）。""" + try: + if self._closing: + return + actor = self._stopped_by.pop(run_id, None) + self._finalizing.add(run_id) + await self._finalize(run_id, "cancelled", "用户取消", {}, 0, + terminal=(EventType.RUN_CANCELLED, {"reason": END_STOPPED, "actor": actor}), + end_reason=END_STOPPED) + finally: + self._finalizing.discard(run_id) + self._tasks.pop(run_id, None) + self._stopped_by.pop(run_id, None) async def _handle_custom(self, run_id: str, chunk: Any) -> None: """节点通过 get_stream_writer 发来的事件。""" diff --git a/backend/tests/test_multiprocess_e2e.py b/backend/tests/test_multiprocess_e2e.py new file mode 100644 index 00000000..09469f96 --- /dev/null +++ b/backend/tests/test_multiprocess_e2e.py @@ -0,0 +1,480 @@ +"""多进程端到端（docs/design/orchestration-round/00-总方案.md §5 W2-b 的验收）：PostgreSQL 上起两个工作进程（uvicorn，A、B）， +指着同一个库、同一个数据目录，用演示模型（Mock）跑： + +1. 两个进程都登记了、只有一个当调度班； +2. 运行由接到请求的那个进程认领；另一个进程上的 WebSocket 收得到实时事件（含不落库的流式增量）；同一个审批决定同时发给两个 + 进程只执行一次；两边的事件流都按 seq 递增、不重不漏，封存核对得上； +3. 在 A 上跑的运行，从 B 停止：照本进程的停止办（run.cancelled 带上是谁停的）； +4. 定时任务到点只触发一次（两个进程都起着调度循环，只有当班的做）； +5. 评测和问数据并发时优先级生效：全站只给 1 个名额，评测的作答运行（系统类，由本测试进程作为第三个登记的工作进程以评测的 + 来源发起——工作进程里发起评测要真的去调模型建图）排着队时，后来的问数据运行（交互类，经 A 的接口发起）先拿到名额； +6. 每日花费上限跨进程不越过：两个进程同时往同一本账上记，记下的合计不超过上限； +7. 强杀 B：租约过期后，B 上在跑的运行被照实标中断（可从断点继续），停在审批上的运行照常能在 A 上批、接着跑完；B 当着调度班的 + 话 A 接班。 + +缺省不跑：要 AGENTLAB_MULTIPROCESS_E2E=1 和 PostgreSQL 测试库（scripts/test-multiprocess.sh 起临时容器、设好这两样）。 +""" +from __future__ import annotations + +import asyncio +import json +import os +import signal +import socket +import subprocess +import sys +import time +from pathlib import Path +from typing import Any + +import pytest + +pytestmark = pytest.mark.skipif( + os.environ.get("AGENTLAB_MULTIPROCESS_E2E") != "1" or not os.environ.get("AGENTLAB_TEST_DATABASE_URL"), + reason="多进程端到端缺省不跑：用 scripts/test-multiprocess.sh（要本机 docker 的 postgres:16）") + +BACKEND = Path(__file__).resolve().parents[1] +#: 工作进程的租约和心跳：调小，强杀之后几秒就能看到接手 +HEARTBEAT_S = 0.5 +LEASE_S = 2.0 + + +def _free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _node(nid: str, ntype: str, **config: Any) -> dict[str, Any]: + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def _chain(*nodes: dict[str, Any]) -> dict[str, Any]: + return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +def _sleep_code(seconds: float) -> dict[str, Any]: + return _node("wait", "code", language="python", code=f"import time\ntime.sleep({seconds})\nprint('ok')") + + +def slow_graph(seconds: float) -> dict[str, Any]: + return _chain(_node("start", "input", fields=[{"name": "question"}]), _sleep_code(seconds), + _node("out", "output", fields=[{"name": "r", "value": "{{ nodes.wait.text }}"}])) + + +APPROVAL_GRAPH = _chain( + _node("start", "input", fields=[{"name": "question"}]), + _sleep_code(1.0), + _node("gate", "human", mode="approve", title="看一眼再接着"), + _node("write", "llm", prompt="用三句话写一段说明"), + _node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.write.text }}"}]), +) + + +class Worker: + def __init__(self, name: str, env: dict[str, str], log_dir: Path) -> None: + self.name = name + self.port = _free_port() + self.base = f"http://127.0.0.1:{self.port}" + self.log = log_dir / f"worker-{name}.log" + self.proc = subprocess.Popen( + [sys.executable, "-m", "uvicorn", "app.main:app", "--host", "127.0.0.1", "--port", str(self.port)], + cwd=str(BACKEND), env=env, stdout=open(self.log, "wb"), stderr=subprocess.STDOUT) + self.owner: str | None = None + + def kill(self) -> None: + if self.proc.poll() is None: + self.proc.send_signal(signal.SIGKILL) + self.proc.wait(timeout=10) + + def stop(self) -> None: + if self.proc.poll() is None: + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(timeout=15) + except subprocess.TimeoutExpired: + self.kill() + + def tail(self) -> str: + try: + return self.log.read_text(encoding="utf-8", errors="replace")[-4000:] + except OSError: + return "" + + +def _worker_env() -> dict[str, str]: + env = {k: v for k, v in os.environ.items() + if k not in ("ANTHROPIC_API_KEY", "ANTHROPIC_BASE_URL", "ANTHROPIC_AUTH_TOKEN", "OPENAI_API_KEY", + "PYTEST_XDIST_WORKER", "PYTEST_CURRENT_TEST")} + env.update({ + "PYTHONPATH": str(BACKEND), + # 应用库、数据目录：和本测试进程同一个（conftest 已经设好） + "AGENTLAB_DATABASE_URL": os.environ["AGENTLAB_DATABASE_URL"], + "AGENTLAB_DATA_DIR": os.environ["AGENTLAB_DATA_DIR"], + "AGENTLAB_DB_POOL_SIZE": "5", + "AGENTLAB_SCHEDULER": "1", + "AGENTLAB_SCHEDULER_TICK_S": "0.5", + "AGENTLAB_WORKER_HEARTBEAT_S": str(HEARTBEAT_S), + "AGENTLAB_WORKER_LEASE_S": str(LEASE_S), + "AGENTLAB_SANDBOX_BACKEND": "local", + "AGENTLAB_AUTH_MODE": "off", + }) + return env + + +@pytest.fixture(scope="module") +def workers(tmp_path_factory): + import httpx + + log_dir = tmp_path_factory.mktemp("multiprocess") + env = _worker_env() + started = [Worker("A", env, log_dir), Worker("B", env, log_dir)] + try: + deadline = time.monotonic() + 90 + for w in started: + while True: + if w.proc.poll() is not None: + raise AssertionError(f"工作进程 {w.name} 没起来：\n{w.tail()}") + try: + if httpx.get(f"{w.base}/api/health", timeout=2).status_code == 200: + break + except httpx.HTTPError: + pass + if time.monotonic() > deadline: + raise AssertionError(f"工作进程 {w.name} 90 秒没起来：\n{w.tail()}") + time.sleep(0.3) + w.owner = httpx.get(f"{w.base}/api/system", timeout=10).json()["cluster"]["me"] + yield {w.name: w for w in started} + finally: + for w in started: + w.stop() + for w in started: + print(f"---- 工作进程 {w.name} 的日志（末尾）----\n{w.tail()}") + + +async def _client(): + import httpx + + return httpx.AsyncClient(timeout=30) + + +async def _until(check, timeout: float = 20.0, what: str = "") -> Any: + deadline = time.monotonic() + timeout + last = None + while time.monotonic() < deadline: + last = await check() + if last: + return last + await asyncio.sleep(0.1) + raise AssertionError(f"等超时了：{what}（最后一次：{last!r}）") + + +async def _run(client, worker: Worker, run_id: str) -> dict[str, Any]: + return (await client.get(f"{worker.base}/api/runs/{run_id}")).json() + + +async def _db(sql: str, *args: Any) -> list[Any]: + from sqlalchemy import text + + from app.db.base import SessionLocal + + async with SessionLocal() as session: + return list((await session.execute(text(sql), _params(args))).all()) + + +def _params(args: tuple[Any, ...]) -> dict[str, Any]: + return {f"p{i}": v for i, v in enumerate(args)} + + +class Stream: + """一条 WebSocket（/api/runs/{id}/stream）收到的全部消息。""" + + def __init__(self, worker: Worker, run_id: str) -> None: + self.url = f"ws://127.0.0.1:{worker.port}/api/runs/{run_id}/stream?after=0" + self.messages: list[dict[str, Any]] = [] + self.task: asyncio.Task[None] | None = None + + async def start(self) -> Stream: + import websockets + + ready = asyncio.Event() + + async def pump() -> None: + async with websockets.connect(self.url) as ws: + ready.set() + async for raw in ws: + msg = json.loads(raw) + self.messages.append(msg) + if msg.get("type") == "stream.end": + return + + self.task = asyncio.create_task(pump()) + await asyncio.wait_for(ready.wait(), timeout=10) + return self + + def types(self) -> list[str]: + return [m.get("type") for m in self.messages] + + def seqs(self) -> list[int]: + """落库事件的 seq（不落库的流式增量另算）。""" + return [int(m["seq"]) for m in self.messages if m.get("seq") and m.get("type") not in ("llm.token", + "llm.thinking.delta")] + + async def ended(self, timeout: float = 30.0) -> None: + assert self.task is not None + await asyncio.wait_for(self.task, timeout=timeout) + + +# -------------------------------------------------------------------------- + + +async def test_1_two_workers_one_leader(workers): + a, b = workers["A"], workers["B"] + async with await _client() as client: + async def both() -> Any: + info = (await client.get(f"{a.base}/api/system")).json()["cluster"] + alive = {w["id"] for w in info["workers"] if w["alive"]} + return info if {a.owner, b.owner} <= alive else None + + info = await _until(both, what="两个进程都登记") + assert info["database"] == "postgresql" + assert info["leader"]["owner"] in (a.owner, b.owner) + other = (await client.get(f"{b.base}/api/system")).json()["cluster"] + assert other["leader"]["owner"] == info["leader"]["owner"], "两边看到的当班是同一个" + + +async def test_2_live_events_across_workers_and_one_decision(workers): + a, b = workers["A"], workers["B"] + async with await _client() as client: + made = (await client.post(f"{a.base}/api/runs", json={"graph": APPROVAL_GRAPH, "input": {"question": "q"}, + "origin": "canvas"})).json() + run_id = made["id"] + # 两边各接一条流（趁它还在跑：停在审批上之后才接的流回放完就结束，要等恢复之后重新接） + on_b = await Stream(b, run_id).start() + on_a = await Stream(a, run_id).start() + owner = await _until(lambda: _owner_while_active(run_id), what="运行被认领") + assert owner == a.owner, "运行由接到请求的进程认领" + + async def paused() -> Any: + return (await _run(client, b, run_id))["status"] == "interrupted" + + await _until(paused, what="停在审批上") + await _until(lambda: _has(on_b, "run.interrupted"), what="B 上的流收到 run.interrupted") + # 同一个决定同时发给两个进程：只执行一次 + decided = await asyncio.gather( + client.post(f"{a.base}/api/runs/{run_id}/resume", json={"response": {"approved": True}}), + client.post(f"{b.base}/api/runs/{run_id}/resume", json={"response": {"approved": True}})) + assert sorted(r.status_code for r in decided)[0] == 200, [r.text for r in decided] + + async def finished() -> Any: + return (await _run(client, a, run_id))["status"] in ("succeeded", "failed") + + await _until(finished, timeout=60, what="运行跑完") + assert (await _run(client, a, run_id))["status"] == "succeeded" + await on_a.ended() + await on_b.ended() + for stream in (on_a, on_b): + assert "run.finished" in stream.types(), stream.types() + seqs = stream.seqs() + assert seqs == sorted(set(seqs)), f"事件流有重复或乱序：{seqs}" + started = [m for m in on_a.messages if m.get("type") == "node.started" and m.get("node_id") == "write"] + assert len(started) == 1, "审批之后的那一步只执行了一次" + # 执行后半段的进程不是谁、另一个进程上的流照样收到了流式增量（不落库，只转给在看的进程） + assert any(m.get("type") == "llm.token" for m in on_a.messages) and \ + any(m.get("type") == "llm.token" for m in on_b.messages), "两边都收到了正文增量" + rows = await _db("SELECT seq, count(*) FROM run_events WHERE run_id = :p0 GROUP BY seq HAVING count(*) > 1", run_id) + assert rows == [], f"库里的事件撞了号：{rows}" + verified = (await client.get(f"{b.base}/api/runs/{run_id}/verify")).json() + assert verified["ok"] is True, verified + + +async def _owner_while_active(run_id: str) -> str | None: + rows = await _db("SELECT owner, status FROM runs WHERE id = :p0", run_id) + if rows and rows[0][1] in ("running", "queued") and rows[0][0]: + return rows[0][0] + return None + + +async def _has(stream: Stream, kind: str) -> bool: + return kind in stream.types() + + +async def test_3_stop_from_the_other_worker(workers): + a, b = workers["A"], workers["B"] + async with await _client() as client: + run_id = (await client.post(f"{a.base}/api/runs", json={"graph": slow_graph(20), "input": {"question": "q"}, + "origin": "canvas"})).json()["id"] + await _until(lambda: _owner_while_active(run_id), what="运行被认领") + t0 = time.monotonic() + stop = await client.post(f"{b.base}/api/runs/{run_id}/cancel", headers={"X-Actor": "e2e-stopper"}) + assert stop.status_code == 200 and stop.json()["status"] == "stopping", stop.text + + async def cancelled() -> Any: + row = await _run(client, b, run_id) + return row if row["status"] == "cancelled" else None + + row = await _until(cancelled, timeout=15, what="停止生效") + assert row["end_reason"] == "stopped" + assert time.monotonic() - t0 < 10, "不是等它跑完才停的" + events = (await client.get(f"{a.base}/api/runs/{run_id}/events")).json() + stopped = [e for e in events if e["type"] == "run.cancelled"] + assert stopped and stopped[-1]["data"].get("actor") == "e2e-stopper", stopped + + +async def test_4_a_schedule_fires_once(workers): + a = workers["A"] + async with await _client() as client: + wf = (await client.post(f"{a.base}/api/workflows", json={"name": "定时验收", "graph": slow_graph(0.1)})).json() + made = await client.post(f"{a.base}/api/schedules", json={ + "name": "每分钟一次", "workflow_id": wf["id"], "cron": "* * * * *", "timezone": "Asia/Shanghai", "run_class": "exploratory", + "input_template": {"question": "定时"}}) + assert made.status_code == 201, made.text + schedule_id = made.json()["id"] + # 拨到「刚过点」：两个进程的调度循环都在转，只有当班的那个做 + from sqlalchemy import text + + from app.db.base import SessionLocal, utcnow + + due = utcnow().replace(microsecond=0) + async with SessionLocal() as session: + await session.execute(text("UPDATE schedules SET next_run_at = :t WHERE id = :i"), {"t": due.replace(tzinfo=None), + "i": schedule_id}) + await session.commit() + + async def fired() -> Any: + rows = await _db("SELECT id FROM schedule_fires WHERE schedule_id = :p0", schedule_id) + return rows or None + + await _until(fired, timeout=15, what="定时触发") + await asyncio.sleep(3) # 再等几拍：不会再发一次 + fires = await _db("SELECT kind, status, run_id FROM schedule_fires WHERE schedule_id = :p0", schedule_id) + runs = await _db("SELECT id FROM runs WHERE schedule_id = :p0", schedule_id) + assert len(fires) == 1 and len(runs) == 1, (fires, runs) + await client.patch(f"{a.base}/api/schedules/{schedule_id}", json={"enabled": False}) + + +async def test_5_chat_goes_before_queued_evaluation_runs(workers, monkeypatch): + """全站 1 个名额：评测的作答运行（系统类）排着两个，问数据的运行（交互类）后来，先拿到名额。""" + from app import cluster + from app.core.config import settings + from app.cluster import slots + from app.cluster.relay import relay + from app.db.base import SessionLocal + from app.db.models import Run + from app.engine.runner import run_manager + + a = workers["A"] + async with await _client() as client: + put = await client.put(f"{a.base}/api/settings", json={"values": {"concurrency": { + "max_total": 1, "classes": {"interactive": {"limit": 1, "priority": 40}, "system": {"limit": 1, "priority": 10}}}}}) + assert put.status_code == 200, put.text + slots.forget_policy() + monkeypatch.setattr(settings, "sandbox_backend", "local") + await cluster.start() # 本测试进程登记成第三个工作进程，替评测发起作答运行 + await relay.start() + await run_manager.setup() + try: + evals = [await run_manager.start(graph=slow_graph(1.5), input_payload={"question": f"评测 {i}"}, + origin_kind="eval", origin={"kind": "eval"}) for i in range(3)] + + async def first_running() -> Any: + async with SessionLocal() as session: + return (await session.get(Run, evals[0].id)).status == "running" + + await _until(first_running, what="第一题的作答运行拿到名额") + await asyncio.sleep(0.3) + chat_id = (await client.post(f"{a.base}/api/runs", json={ + "graph": slow_graph(0.2), "input": {"question": "问数据"}, "origin": "chat"})).json()["id"] + + async def all_done() -> Any: + async with SessionLocal() as session: + rows = [await session.get(Run, rid, populate_existing=True) for rid in [*[e.id for e in evals], chat_id]] + return rows if all(r.status in ("succeeded", "failed") for r in rows) else None + + rows = await _until(all_done, timeout=60, what="都跑完") + assert all(r.status == "succeeded" for r in rows), [(r.id, r.status, r.error) for r in rows] + e1, e2, e3, chat = rows + assert e2.created_at < chat.created_at and e3.created_at < chat.created_at, "评测的两题先来、排着队" + assert e1.started_at < chat.started_at < e2.started_at < e3.started_at, \ + [(r.origin_kind, r.created_at, r.started_at) for r in rows] + finally: + await run_manager.shutdown() + await relay.stop() + await cluster.stop() + await client.put(f"{a.base}/api/settings", json={"values": {"concurrency": slots.defaults()}}) + + +def test_6_the_daily_cap_holds_across_processes(workers): + """两个进程同时往评测那本账上各记 20 次、每次 0.1 美元，上限 1 美元：合计记下的正好 10 次、不超过上限。""" + script = ( + "import asyncio, json\n" + "from app.engine.spend import Ledger\n" + "async def main():\n" + " book = Ledger('e2e_spend')\n" + " got = await asyncio.gather(*(book.charge(0.1, limit=1.0, priced=True) for _ in range(20)))\n" + " print(json.dumps(sum(1 for d in got if d)))\n" + "asyncio.run(main())\n") + env = {**_worker_env(), "AGENTLAB_SCHEDULER": "0"} + procs = [subprocess.Popen([sys.executable, "-c", script], cwd=str(BACKEND), env=env, stdout=subprocess.PIPE, + stderr=subprocess.PIPE) for _ in range(2)] + outs = [p.communicate(timeout=120) for p in procs] + assert all(p.returncode == 0 for p in procs), [o[1].decode()[-2000:] for o in outs] + charged = sum(int(o[0].decode().strip().splitlines()[-1]) for o in outs) + assert charged == 10, f"两个进程合计记下 {charged} 次（上限只容得下 10 次）" + total = asyncio.run(_spent()) + assert total <= 1.0 + 1e-9, total + + +async def _spent() -> float: + from app.engine.spend import Ledger + + return float((await Ledger("e2e_spend").daily())["usd"]) + + +async def test_7_kill_a_worker(workers): + a, b = workers["A"], workers["B"] + async with await _client() as client: + leader_before = (await client.get(f"{a.base}/api/system")).json()["cluster"]["leader"]["owner"] + # B 上：一次在跑（会被照实标中断），一次停在审批上（不属于任何进程，B 没了照样能在 A 上批） + busy = (await client.post(f"{b.base}/api/runs", json={"graph": slow_graph(30), "input": {"question": "q"}, + "origin": "canvas"})).json()["id"] + paused = (await client.post(f"{b.base}/api/runs", json={"graph": APPROVAL_GRAPH, "input": {"question": "q"}, + "origin": "canvas"})).json()["id"] + await _until(lambda: _owner_while_active(busy), what="B 认领了运行") + + async def waiting() -> Any: + return (await _run(client, a, paused))["status"] == "interrupted" + + await _until(waiting, what="停在审批上") + b.kill() + t0 = time.monotonic() + + async def taken_over() -> Any: + row = await _run(client, a, busy) + return row if row["status"] == "interrupted" else None + + row = await _until(taken_over, timeout=LEASE_S * 6 + 10, what="租约过期后被接手") + assert "可从断点继续运行" in (row["error"] or "") + events = (await client.get(f"{a.base}/api/runs/{busy}/events")).json() + assert any(e["type"] == "log" and e["data"].get("code") == "server_shutdown" for e in events) + print(f"强杀 B 之后 {time.monotonic() - t0:.1f} 秒被接手") + info = (await client.get(f"{a.base}/api/system")).json()["cluster"] + assert b.owner not in {w["id"] for w in info["workers"] if w["alive"]} + if leader_before == b.owner: + async def a_leads() -> Any: + got = (await client.get(f"{a.base}/api/system")).json()["cluster"]["leader"] + return got["owner"] == a.owner + + await _until(a_leads, timeout=LEASE_S * 6, what="A 接班") + # 停在审批上的那一次：在 A 上批，接着跑完 + ok = await client.post(f"{a.base}/api/runs/{paused}/resume", json={"response": {"approved": True}}) + assert ok.status_code == 200, ok.text + + async def done() -> Any: + return (await _run(client, a, paused))["status"] == "succeeded" + + await _until(done, timeout=60, what="批了之后在 A 上跑完") + # 被标中断的那一次：在 A 上能接着跑（这里接上就停掉，不等它睡完） + cont = await client.post(f"{a.base}/api/runs/{busy}/continue", json={}) + assert cont.status_code == 200, cont.text + await client.post(f"{a.base}/api/runs/{busy}/cancel") diff --git a/backend/tests/test_p3_api_errors.py b/backend/tests/test_p3_api_errors.py index 6fb872de..1b3466ae 100644 --- a/backend/tests/test_p3_api_errors.py +++ b/backend/tests/test_p3_api_errors.py @@ -136,6 +136,7 @@ def held_guard(store): table_versions._release_guard() +@pytest.mark.sqlite_only # 文件锁守卫只在 SQLite（单进程）时有：PostgreSQL 可以多进程，版本存储靠跨进程的锁（app/cluster/mutex） async def test_read_only_process_refuses_writes_with_503_and_keeps_reads(client, held_guard): name = unique() base = await first_import(client, name, seed=505) @@ -170,6 +171,7 @@ async def test_read_only_process_refuses_writes_with_503_and_keeps_reads(client, assert await table_versions.expire_stagings(session) == [] +@pytest.mark.sqlite_only # 文件锁守卫只在 SQLite（单进程）时有：PostgreSQL 可以多进程，版本存储靠跨进程的锁（app/cluster/mutex） async def test_read_only_process_refuses_activate_and_commit_keeps_the_trial(client, held_guard): """启用旧版本 503；提交 503 时不消耗试运行库（守卫拿回来之后照常提交）。""" name = unique() @@ -196,6 +198,7 @@ async def test_read_only_process_refuses_activate_and_commit_keeps_the_trial(cli assert resp.json()["parts"] == 3 +@pytest.mark.sqlite_only # 文件锁守卫只在 SQLite（单进程）时有：PostgreSQL 可以多进程，版本存储靠跨进程的锁（app/cluster/mutex） async def test_read_only_process_refuses_version_and_edit_writes_before_doing_any_work(client, held_guard, store, monkeypatch): """只读进程：移除一期（要物化）、作废接受（累积模式，要物化）、edits/apply、edits/undo、POST /{id}/redraft 一律 503， @@ -253,6 +256,7 @@ async def test_read_only_process_refuses_version_and_edit_writes_before_doing_an assert len((await db_get(ImportStaging, applied["id"])).edits) == 1, "撤销没有生效" +@pytest.mark.sqlite_only # 文件锁守卫只在 SQLite（单进程）时有：PostgreSQL 可以多进程，版本存储靠跨进程的锁（app/cluster/mutex） async def test_read_only_process_still_deletes_a_manual_source(client, held_guard): """手工登记的源没有导入记录、配方、暂存区和隔离文件，不碰版本存储：只读进程里照常删除（9.1 只要求上传源 503）。""" sid = "manual-" + unique("m") diff --git a/backend/tests/test_recipe_acceptance_p3.py b/backend/tests/test_recipe_acceptance_p3.py index a502564c..84ad61f0 100644 --- a/backend/tests/test_recipe_acceptance_p3.py +++ b/backend/tests/test_recipe_acceptance_p3.py @@ -1683,6 +1683,7 @@ sys.stdin.read() """ +@pytest.mark.sqlite_only # 文件锁守卫只在 SQLite（单进程）时有：PostgreSQL 可以多进程，版本存储靠跨进程的锁（app/cluster/mutex） async def test_l5_another_process_holding_the_store_guard(client, store): """L5：子进程先占住 .store.lock → 本进程 acquire_store_guard() 返回 False；本进程调 startup：事先放好的临时文件和孤儿 构建库一个都没删；/upload、stage、trial、commit、activate 返回 503 store_unavailable，GET 接口照常；过期处理不抛； diff --git a/backend/tests/test_table_versions_p3.py b/backend/tests/test_table_versions_p3.py index 13494004..444e1166 100644 --- a/backend/tests/test_table_versions_p3.py +++ b/backend/tests/test_table_versions_p3.py @@ -1162,6 +1162,7 @@ async def test_guard_is_idempotent_in_one_process(store, guard_reset): await table_versions.startup(session) +@pytest.mark.sqlite_only # 文件锁守卫只在 SQLite（单进程）时有：PostgreSQL 可以多进程，版本存储靠跨进程的锁（app/cluster/mutex） async def test_second_process_skips_startup_cleanup_and_refuses_writes(store, guard_reset, monkeypatch): """L5：子进程先占住守卫，本进程启动时一个文件都不删、不回收，写入一律 StoreUnavailable；子进程退出后恢复。""" name = unique() @@ -1243,6 +1244,7 @@ async def test_second_process_skips_startup_cleanup_and_refuses_writes(store, gu assert not half.exists() and not orphan.exists() and not orphan_raw.exists() +@pytest.mark.sqlite_only # 文件锁守卫只在 SQLite（单进程）时有：PostgreSQL 可以多进程，版本存储靠跨进程的锁（app/cluster/mutex） async def test_guard_state_does_not_leak_into_another_data_dir(store, guard_reset, monkeypatch, tmp_path): proc = hold_guard() try: diff --git a/scripts/test-multiprocess.sh b/scripts/test-multiprocess.sh new file mode 100755 index 00000000..1e5652f4 --- /dev/null +++ b/scripts/test-multiprocess.sh @@ -0,0 +1,51 @@ +#!/usr/bin/env bash +# 多进程端到端验收（docs/design/orchestration-round §5 W2-b）：本机 docker 起一个临时的 postgres:16 容器（只绑 127.0.0.1 的随机 +# 端口、数据放内存里，跑完删掉，同 test-postgres.sh），在上面起两个工作进程（uvicorn，指着同一个库、同一个数据目录），用演示模型 +# 跑 backend/tests/test_multiprocess_e2e.py：运行由任一进程认领、另一个进程上的订阅收得到实时事件、同一个审批决定只执行一次、 +# 跨进程停止、定时只触发一次、评测和问数据并发时优先级生效、每日花费上限跨进程不越过、强杀一个进程后租约过期被接手或照实标中断。 +# +# scripts/test-multiprocess.sh 跑一遍（工作进程的日志在失败时打出来） +# scripts/test-multiprocess.sh -k kill 参数原样交给 pytest +# PYTHON=/path/to/python scripts/test-multiprocess.sh 指定装着后端依赖（含 [postgres]、[dev]）的 Python，缺省 python +# AGENTLAB_PG_IMAGE=postgres:16 … 换镜像（要本机已经有，这里不拉） +# +# 作为 pytest 的一组：tests/test_multiprocess_e2e.py 缺省跳过，要 AGENTLAB_MULTIPROCESS_E2E=1 和 AGENTLAB_TEST_DATABASE_URL +# （这个脚本设好两样）。 +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +PY="${PYTHON:-python}" +IMAGE="${AGENTLAB_PG_IMAGE:-postgres:16}" +NAME="agentlab-mptest-$$" +USER_NAME="agentlab" +PASS="mptest-only" +DB="agentlab_test" + +command -v docker >/dev/null || { echo "缺少 docker"; exit 1; } +docker image inspect "$IMAGE" >/dev/null 2>&1 || { echo "本机没有镜像 ${IMAGE}（这个脚本不拉镜像）：先 docker pull ${IMAGE}"; exit 1; } + +cleanup() { docker rm -f "$NAME" >/dev/null 2>&1 || true; } +trap cleanup EXIT INT TERM + +docker run -d --name "$NAME" -e POSTGRES_USER="$USER_NAME" -e POSTGRES_PASSWORD="$PASS" -e POSTGRES_DB="$DB" \ + -e POSTGRES_HOST_AUTH_METHOD=trust \ + -p 127.0.0.1::5432 --tmpfs /var/lib/postgresql/data:rw "$IMAGE" \ + -c fsync=off -c synchronous_commit=off -c full_page_writes=off -c max_connections=300 \ + >/dev/null + +for _ in $(seq 1 60); do + if docker exec "$NAME" pg_isready -U "$USER_NAME" -d "$DB" -h 127.0.0.1 >/dev/null 2>&1; then break; fi + sleep 0.5 +done +docker exec "$NAME" pg_isready -U "$USER_NAME" -d "$DB" -h 127.0.0.1 >/dev/null || { echo "PostgreSQL 没起来"; docker logs "$NAME" | tail -20; exit 1; } +PORT="$(docker port "$NAME" 5432/tcp | head -1 | sed 's/.*://')" + +export AGENTLAB_TEST_DATABASE_URL="postgresql+asyncpg://$USER_NAME:$PASS@127.0.0.1:$PORT/$DB" +export AGENTLAB_MULTIPROCESS_E2E=1 +echo "测试库：127.0.0.1:${PORT}/${DB}（容器 ${NAME}，结束后删除）" + +cd "$ROOT/backend" +args=("$@") +# 不并行（-p no:xdist）：工作进程、端口、库都是这一组共用的；-s 让工作进程的日志在失败时看得到 +env -u ANTHROPIC_API_KEY -u ANTHROPIC_BASE_URL -u ANTHROPIC_AUTH_TOKEN -u OPENAI_API_KEY \ + "$PY" -m pytest tests/test_multiprocess_e2e.py -p no:xdist -p no:randomly -rA "${args[@]}"