From 35ab097b459a11cbc1d6dd59c8170c0e1981a862 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 13:15:56 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=8D=E6=A0=B8=20R-b1=EF=BC=88=E5=8D=A1?= =?UTF-8?q?=E4=BD=8F=E4=BD=86=E6=B2=A1=E6=AD=BB=E7=9A=84=E8=BF=9B=E7=A8=8B?= =?UTF-8?q?=E8=A2=AB=E5=BD=93=E6=88=90=E6=B2=A1=E4=BA=86=EF=BC=89=EF=BC=9A?= =?UTF-8?q?=E7=BB=AD=E7=A7=9F=E7=BA=A6=E6=8C=AA=E5=88=B0=E5=8D=95=E7=8B=AC?= =?UTF-8?q?=E7=9A=84=E4=BB=BB=E5=8A=A1=E3=80=81=E4=B8=93=E7=94=A8=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=EF=BC=88=E4=B8=8D=E8=B5=B0=E8=BF=9E=E6=8E=A5=E6=B1=A0?= =?UTF-8?q?=EF=BC=8C=E4=B8=8D=E8=A2=AB=E5=90=8C=E4=B8=80=E6=8B=8D=E5=88=AB?= =?UTF-8?q?=E7=9A=84=E6=B4=BB=E6=8B=96=E4=BD=8F=EF=BC=89=EF=BC=9B=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E9=97=B4=E8=AF=B7=E6=B1=82=E5=8F=A6=E8=B5=B7=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E5=8A=9E=EF=BC=8C=E4=B8=8D=E5=9C=A8=E5=BF=83=E8=B7=B3?= =?UTF-8?q?=E9=87=8C=E7=AD=89=E6=94=B6=E5=B0=BE=EF=BC=9B=E6=96=B0=E5=A2=9E?= =?UTF-8?q?=20cluster/fence=20=E8=87=AA=E6=88=91=E9=9A=94=E7=A6=BB?= =?UTF-8?q?=E2=80=94=E2=80=94=E7=A6=BB=E4=B8=8A=E6=AC=A1=E7=BB=AD=E4=B8=8A?= =?UTF-8?q?=E7=A7=9F=E7=BA=A6=E8=B6=85=E8=BF=87=E4=B8=80=E4=B8=AA=E7=A7=9F?= =?UTF-8?q?=E7=BA=A6=E6=97=B6=E9=95=BF=EF=BC=8C=E6=88=96=E8=80=85=E8=BF=99?= =?UTF-8?q?=E6=AE=B5=E6=89=A7=E8=A1=8C=E5=B1=9E=E4=BA=8E=E4=B8=8A=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=E9=9A=94=E7=A6=BB=E4=B9=8B=E5=89=8D=EF=BC=8C=E6=89=A7?= =?UTF-8?q?=E8=A1=8C=E5=9C=A8=E5=86=99=E4=BA=8B=E4=BB=B6=E3=80=81=E5=86=99?= =?UTF-8?q?=E6=96=AD=E7=82=B9=E3=80=81=E5=BC=80=E5=AE=A1=E6=89=B9=E3=80=81?= =?UTF-8?q?=E5=8F=91=E5=A4=96=E9=83=A8=E5=86=99=E5=85=A5=E3=80=81=E5=90=8E?= =?UTF-8?q?=E5=8F=B0=E4=BB=BB=E5=8A=A1=E5=86=99=E5=BA=93=E4=B9=8B=E5=89=8D?= =?UTF-8?q?=E5=B0=B1=E5=81=9C=E6=89=8B=EF=BC=88Fenced=20=E6=98=AF=20BaseEx?= =?UTF-8?q?ception=EF=BC=8C=E8=8A=82=E7=82=B9=E9=87=8C=E7=9A=84=20except?= =?UTF-8?q?=20Exception=20=E6=8B=A6=E4=B8=8D=E4=BD=8F=EF=BC=89=EF=BC=8C?= =?UTF-8?q?=E7=BB=AD=E7=A7=9F=E6=97=B6=E5=8F=91=E7=8E=B0=E7=BB=AD=E6=99=9A?= =?UTF-8?q?=E4=BA=86=E5=B0=B1=E9=9A=94=E7=A6=BB=EF=BC=9A=E5=81=9C=E6=8E=89?= =?UTF-8?q?=E6=89=8B=E4=B8=8A=E7=9A=84=E8=BF=90=E8=A1=8C=E5=92=8C=E5=90=8E?= =?UTF-8?q?=E5=8F=B0=E4=BB=BB=E5=8A=A1=E3=80=81=E4=B8=8D=E5=86=99=E7=BB=88?= =?UTF-8?q?=E6=80=81=EF=BC=8C=E6=8D=A2=E4=B8=80=E4=B8=AA=E8=BA=AB=E4=BB=BD?= =?UTF-8?q?=E9=87=8D=E6=96=B0=E7=99=BB=E8=AE=B0=EF=BC=9B=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E6=94=B6=E5=B0=BE=E5=85=88=E5=8D=A0=E4=BD=8F=E9=82=A3=E4=B8=80?= =?UTF-8?q?=E8=A1=8C=E3=80=81=E5=B1=9E=E4=B8=BB=E4=B8=8D=E6=98=AF=E8=87=AA?= =?UTF-8?q?=E5=B7=B1=E5=B0=B1=E4=B8=8D=E5=86=99=EF=BC=9B=E5=88=AB=E7=9A=84?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E6=8C=89=E7=A7=9F=E7=BA=A6=E8=BF=87=E6=9C=9F?= =?UTF-8?q?=E5=86=8D=E5=A4=9A=E7=AD=89=E4=B8=80=E4=B8=AA=E7=A7=9F=E7=BA=A6?= =?UTF-8?q?=E6=97=B6=E9=95=BF=E6=89=8D=E7=AE=97=E4=B8=8D=E5=9C=A8=E4=BA=86?= =?UTF-8?q?=EF=BC=88=E4=B8=A4=E8=BE=B9=E4=B8=AD=E9=97=B4=E7=95=99=E4=B8=80?= =?UTF-8?q?=E6=AE=B5=EF=BC=89=EF=BC=9Brun=5Fevents=20=E7=9A=84=20(run=5Fid?= =?UTF-8?q?,=20seq)=20=E6=94=B9=E6=88=90=E5=94=AF=E4=B8=80=E7=B4=A2?= =?UTF-8?q?=E5=BC=95=EF=BC=88=E5=AD=98=E9=87=8F=E5=BA=93=E5=85=88=E6=95=B0?= =?UTF-8?q?=E9=87=8D=E5=A4=8D=EF=BC=8C=E6=9C=89=E5=B0=B1=E7=85=A7=E5=AE=9E?= =?UTF-8?q?=E8=AE=B0=E6=97=A5=E5=BF=97=E3=80=81=E8=BF=99=E4=B8=80=E6=AC=A1?= =?UTF-8?q?=E4=B8=8D=E5=BB=BA=EF=BC=8C=E4=B8=8D=E8=AE=A9=E5=90=AF=E5=8A=A8?= =?UTF-8?q?=E5=A4=B1=E8=B4=A5=EF=BC=89=E3=80=82=E8=A1=A5=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=EF=BC=9A=E5=8D=A1=E4=BD=8F=E6=9C=9F=E9=97=B4=E8=A2=AB=E6=8E=A5?= =?UTF-8?q?=E6=89=8B=E4=B9=8B=E5=90=8E=E4=B8=8D=E7=9B=96=E7=BB=88=E6=80=81?= =?UTF-8?q?=E3=80=81=E4=B8=8D=E5=86=99=E4=BA=8B=E4=BB=B6=E3=80=81=E5=BA=8F?= =?UTF-8?q?=E5=8F=B7=E4=B8=8D=E9=87=8D=EF=BC=9B=E8=A2=AB=E6=8E=A5=E6=89=8B?= =?UTF-8?q?=E7=9A=84=E9=82=A3=E6=AC=A1=E5=9C=A8=E5=88=AB=E5=A4=84=E6=8E=A5?= =?UTF-8?q?=E7=9D=80=E8=B7=91=E5=AE=8C=E5=8F=AA=E7=BB=93=E6=9D=9F=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=E3=80=81=E5=90=8E=E9=9D=A2=E9=82=A3=E4=B8=80=E6=AD=A5?= =?UTF-8?q?=E5=8F=AA=E8=B7=91=E4=B8=80=E6=AC=A1=EF=BC=9B=E6=94=B6=E5=B0=BE?= =?UTF-8?q?=E6=8C=89=E5=B1=9E=E4=B8=BB=E5=86=99=EF=BC=9B=E6=92=9E=E5=8F=B7?= =?UTF-8?q?=E8=A2=AB=E5=BA=93=E6=8B=92=E7=BB=9D=EF=BC=9B=E5=A4=9A=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E7=AB=AF=E5=88=B0=E7=AB=AF=E5=8A=A0=206b=E3=80=816c?= =?UTF-8?q?=EF=BC=88=E7=9C=9F=20SIGSTOP=20/=20SIGCONT=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/cluster/fence.py | 116 +++++++++++++++++++ backend/app/cluster/identity.py | 10 +- backend/app/cluster/leader.py | 9 ++ backend/app/cluster/requests.py | 28 ++++- backend/app/cluster/service.py | 143 ++++++++++++++++++++---- backend/app/cluster/workers.py | 7 +- backend/app/core/job_runtime.py | 24 +++- backend/app/db/base.py | 24 ++++ backend/app/db/models.py | 4 +- backend/app/engine/runner.py | 69 ++++++++++-- backend/app/process/actions.py | 4 + backend/tests/test_cluster_fence.py | 149 +++++++++++++++++++++++++ backend/tests/test_multiprocess_e2e.py | 99 +++++++++++++++- 13 files changed, 643 insertions(+), 43 deletions(-) create mode 100644 backend/app/cluster/fence.py create mode 100644 backend/tests/test_cluster_fence.py diff --git a/backend/app/cluster/fence.py b/backend/app/cluster/fence.py new file mode 100644 index 00000000..60da513b --- /dev/null +++ b/backend/app/cluster/fence.py @@ -0,0 +1,116 @@ +"""自我隔离（fencing）：本进程续不上租约的时候，当作自己认领的行可能已经被别的进程接手，停下手上的执行、不再写。 + +**为什么。** 进程卡住了但没死（SIGSTOP、虚拟机暂停、事件循环被长同步代码卡住、库一时连不上）：别的进程看它的租约过期了，把它在跑 +的运行标中断、交给别人继续运行。它一缓过来，手上的执行接着往下走：照样写事件（和接手的那一边撞号）、照样写终态（把接手的结果 +盖掉）。所以： + +- 续租约在单独的任务、单独的连接上做（cluster/service.py），不被同一拍别的活拖住；最近一次续上的时刻记在这里（renewed）。 +- 本进程的执行在动手写之前问一句 check()：离上次续上已经超过一个租约时长（别的进程按两个租约时长才接手，中间留着一段），或者这段 + 执行是在上一次隔离之前开始的（世代不对），就抛 Fenced——写事件、写断点、写终态、写待办和动作、发外部写入之前都问。 +- 一旦发现续晚了（trip）：世代加一，手上的执行全部停掉（停的时候不写终态，那些行交给接手的进程按「属主不在了」处理），换一个新的 + 身份重新登记，之后发起的照常。 + +Fenced 是 BaseException：节点、工具里常见的 except Exception 拦不住它，一路退到执行的最外层。没起协调（测试、脚本）时什么都不查。 +""" +from __future__ import annotations + +import contextvars +import logging +import time +from collections.abc import Callable +from typing import Any + +logger = logging.getLogger(__name__) + + +class Fenced(BaseException): + """本进程续不上租约了（或者这段执行属于上一次隔离之前）：不许再写。""" + + +_generation = 0 +_renewed: float | None = None +_active = False +_handlers: list[Callable[[], Any]] = [] +_bound: contextvars.ContextVar[int | None] = contextvars.ContextVar("agentlab_fence_generation", default=None) + + +def activate(renewed_at: float | None = None) -> None: + """起协调、续上了第一次租约时（PostgreSQL；SQLite 只有一个进程，不用隔离）。""" + global _active, _renewed + _active = True + _renewed = time.monotonic() if renewed_at is None else renewed_at + + +def deactivate() -> None: + global _active, _renewed + _active, _renewed = False, None + + +def renewed(at: float | None = None) -> None: + """续上了一次租约。""" + global _renewed + _renewed = time.monotonic() if at is None else at + + +def generation() -> int: + return _generation + + +def bind() -> None: + """这段执行属于现在这一世代（运行的一段执行、后台任务、问数据的一轮开始时调）。""" + _bound.set(_generation) + + +def stale(now: float | None = None) -> bool: + """离上次续上租约是不是已经超过一个租约时长。""" + if not _active or _renewed is None: + return False + from app.cluster.workers import lease_s + + return (time.monotonic() if now is None else now) - _renewed > lease_s() + + +def fenced_here() -> bool: + """当前这段执行还能不能写：世代不对（隔离之前开始的）或者租约续晚了。""" + if not _active: + return False + bound = _bound.get() + return (bound is not None and bound != _generation) or stale() + + +def check() -> None: + """写之前问一句：不许写就抛 Fenced（续晚了的话顺手隔离）。""" + if not _active: + return + if stale(): + trip("执行要写的时候发现租约已经续晚了") + raise Fenced() + bound = _bound.get() + if bound is not None and bound != _generation: + raise Fenced() + + +def on_trip(handler: Callable[[], Any]) -> None: + if handler not in _handlers: + _handlers.append(handler) + + +def trip(why: str) -> bool: + """隔离：世代加一，叫各处停掉手上的执行（cluster/service 再换身份重新登记）。已经在这一世代隔离过的不重复。交回这一次有没有动手。""" + global _generation, _renewed + if not _active or _renewed is None: + return False + _generation += 1 + _renewed = None + logger.error("本进程的租约续晚了（%s）：当作认领的运行、问数据、后台任务可能已被别的进程接手，停掉手上的执行、不写终态，" + "换一个身份重新登记", why) + for handler in list(_handlers): + try: + handler() + except Exception: # noqa: BLE001 + logger.exception("隔离时停执行出错") + return True + + +__all__ = ["Fenced", "activate", "bind", "check", "deactivate", "fenced_here", "generation", "on_trip", "renewed", + "stale", "trip"] diff --git a/backend/app/cluster/identity.py b/backend/app/cluster/identity.py index 1a0d6695..7a554012 100644 --- a/backend/app/cluster/identity.py +++ b/backend/app/cluster/identity.py @@ -33,6 +33,14 @@ def owner() -> str: return OWNER +def regenerate() -> str: + """换一个身份（隔离之后重新登记用，cluster/fence.py）：同一个 pid、新的随机串。之前认领的行算「属主不在了」。""" + global OWNER, _PID + _PID = os.getpid() + OWNER = _make(_PID) + return OWNER + + def host() -> str: return _HOST @@ -45,4 +53,4 @@ def describe(value: str | None) -> str: return f"{parts[0]} 上的进程 {parts[1]}" if len(parts) >= 2 else value -__all__ = ["OWNER", "describe", "host", "owner"] +__all__ = ["OWNER", "describe", "host", "owner", "regenerate"] diff --git a/backend/app/cluster/leader.py b/backend/app/cluster/leader.py index ab2d800b..fe169232 100644 --- a/backend/app/cluster/leader.py +++ b/backend/app/cluster/leader.py @@ -32,10 +32,16 @@ class Leadership: self.until: datetime | None = None self.epoch = 0 + def single(self) -> None: + """只有本进程（SQLite）：直接算当班，不写库。""" + self.until = datetime.max.replace(tzinfo=None) + def held(self, now: datetime | None = None) -> bool: """本地看：还在当班（租约离到期还有余量）。""" if self.until is None: return False + if self.until.tzinfo is None: + return True # single()：只有本进程 # 余量不超过租约的四分之一：租约调得很短（验收脚本）时也留得出当班的时间 margin = min(MARGIN_S, lease_s() / 4) return (now or utcnow()) + timedelta(seconds=margin) < self.until @@ -85,7 +91,10 @@ class Leadership: if self.until is None: return + single = self.until.tzinfo is None self.until = None + if single: + return async with SessionLocal() as session: await session.execute(update(ClusterLease).where( ClusterLease.name == self.name, ClusterLease.owner == identity.owner()) diff --git a/backend/app/cluster/requests.py b/backend/app/cluster/requests.py index 09332874..34fb7e8f 100644 --- a/backend/app/cluster/requests.py +++ b/backend/app/cluster/requests.py @@ -80,14 +80,32 @@ async def _drain() -> int: if fn is None: logger.warning("收到不认识的进程间请求 %s/%s，不办", row.kind, row.target) continue - try: - await fn(row.target_id, row.actor, dict(row.payload or {})) - done += 1 - except Exception: # noqa: BLE001 - logger.exception("办进程间请求 %s/%s %s 出错", row.kind, row.target, row.target_id) + # 各办各的、另起任务：停止要等执行收尾（最多十来秒），不能拖住心跳和后面几件 + task = asyncio.ensure_future(_handle(fn, row.kind, row.target, row.target_id, row.actor, dict(row.payload or {}))) + _running.add(task) + task.add_done_callback(_running.discard) + done += 1 return done +#: 正在办的请求（另起的任务要有人拿着引用） +_running: set[asyncio.Future[Any]] = set() + + +async def _handle(fn: Callable[..., Awaitable[Any]], kind: str, target: str, target_id: str, actor: str | None, + payload: dict[str, Any]) -> None: + try: + await fn(target_id, actor, payload) + except Exception: # noqa: BLE001 + logger.exception("办进程间请求 %s/%s %s 出错", kind, target, target_id) + + +async def settle(timeout: float = 5.0) -> None: + """等手上正在办的请求办完（测试、关停用）。""" + if _running: + await asyncio.wait(list(_running), timeout=timeout) + + async def discard_for(owners: set[str]) -> int: """收件人已经不在了的请求删掉（接手的进程做：那些行按属主不在了的规矩处理）。""" from app.db.models_cluster import ClusterRequest diff --git a/backend/app/cluster/service.py b/backend/app/cluster/service.py index 35b52029..ff9cf7b3 100644 --- a/backend/app/cluster/service.py +++ b/backend/app/cluster/service.py @@ -1,7 +1,11 @@ """本进程在协调里的那一份：启动时登记、起心跳，关停时交班、删登记。 -心跳每一拍（AGENTLAB_WORKER_HEARTBEAT_S，缺省 5 秒）： -1. 续自己的租约（登记不见了就重新登记）； +续租约在单独的任务、单独的连接上（不走连接池，不被同一拍别的活拖住）；续晚了（离上次续上超过一个租约时长）就隔离：停掉手上的 +执行、不写终态，换一个身份重新登记（cluster/fence.py）。SQLite 只有一个进程：登记只写一次、不续租、本进程直接当班、接手只在 +启动时做，心跳只做本进程自己攒着的活。 + +心跳每一拍（AGENTLAB_WORKER_HEARTBEAT_S，缺省 5 秒；PostgreSQL）： +1. （续租约另有任务）； 2. 刷新「别的进程谁还活着」； 3. 取走发给本进程的请求照办（停止、往运行里记事件，cluster/requests.py）； 4. 本进程自己的周期活（on_beat 登记的：访问密钥的请求记录写库……——这些是每个进程攒在自己内存里的，只能自己写）； @@ -52,36 +56,48 @@ def leading() -> bool: class _Service: def __init__(self) -> None: self.task: asyncio.Task[None] | None = None + self.lease_task: asyncio.Task[None] | None = None self.wake: asyncio.Event | None = None self.reap_now = False self.last_reap = 0.0 self.others: frozenset[str] = frozenset() self.started = False + #: SQLite：守卫已经确认只有本进程——登记只写一次、不续租、本进程直接当班、接手只在启动时做（不为协调多写库） + self.single = False + self._lease_conn: Any = None async def start(self) -> None: """登记、开始收发通知、起心跳。SQLite 时先取单进程的守卫（拿不到抛 guard.SingleProcessOnly）。""" if self.started: return guard.acquire() - exclusive = settings.db_dialect == "sqlite" - await workers.register(exclusive=exclusive) + self.single = settings.db_dialect == "sqlite" + await workers.register(exclusive=self.single) live = await workers.liveness() state.remember(live) self.others = live.others await signals.start() signals.on("q", self._on_request) signals.on("gone", self._on_gone) - if settings.db_dialect == "postgresql": + if not self.single: await signals.ready(timeout=5.0) self.wake = asyncio.Event() self.started = True - from app.cluster import caches + from app.cluster import caches, fence await caches.start() - await self._lead() - self.task = asyncio.create_task(self._loop(), name="agentlab-cluster-heartbeat") + if self.single: + # 只有本进程：当班不用争（不写库），心跳只做本进程自己攒着的活（访问密钥的请求记录） + leader.scheduler.single() + self.task = asyncio.create_task(self._light_loop(), name="agentlab-cluster-beats") + else: + await self._lead() + fence.activate() + fence.on_trip(self._on_fenced) + self.lease_task = asyncio.create_task(self._lease_loop(), name="agentlab-cluster-lease") + self.task = asyncio.create_task(self._loop(), name="agentlab-cluster-heartbeat") logger.info("工作进程 %s 已登记（%s，租约 %g 秒、心跳 %g 秒；现在另有 %d 个进程）", identity.describe(identity.owner()), - "PostgreSQL，可以多进程" if settings.db_dialect == "postgresql" else "SQLite，单进程", + "SQLite，单进程" if self.single else "PostgreSQL，可以多进程", workers.lease_s(), workers.heartbeat_s(), len(live.others)) def _on_request(self, parts: list[str]) -> None: @@ -93,6 +109,75 @@ class _Service: if self.wake is not None: self.wake.set() + # ---------------- 续租约：单独的任务、单独的连接 ---------------- + + async def _lease_conn_open(self) -> Any: + import asyncpg + + from app.cluster.signal import listen_dsn + from app.db.base import APPLICATION_NAME + + if self._lease_conn is None or self._lease_conn.is_closed(): + self._lease_conn = await asyncpg.connect(listen_dsn(), server_settings={"application_name": APPLICATION_NAME}, + timeout=max(1.0, workers.heartbeat_s())) + return self._lease_conn + + async def renew(self) -> bool: + """续一拍租约（在专用连接上，不走连接池：池子被占满时取连接要等，等得比租约还长）。续晚了（离上次续上超过一个租约时长） + 就不续这个身份了：隔离（cluster/fence.py），换身份重新登记。交回续没续上。""" + from datetime import timedelta + + from app.cluster import fence + from app.db.base import utcnow + + if fence.stale(): + fence.trip("续租约的时候发现已经续晚了") + return False + now = utcnow() + until = now + timedelta(seconds=workers.lease_s()) + try: + conn = await self._lease_conn_open() + done = await asyncio.wait_for(conn.execute( + "UPDATE cluster_workers SET heartbeat_at = $2, lease_until = $3 WHERE id = $1", + identity.owner(), now.replace(tzinfo=None), until.replace(tzinfo=None)), timeout=workers.heartbeat_s()) + except Exception: # noqa: BLE001 - 库一时连不上：这一拍没续上，下一拍再试（续晚了的话下一拍隔离） + logger.warning("续租约没续上，下一拍再试", exc_info=True) + with contextlib.suppress(Exception): + if self._lease_conn is not None: + await self._lease_conn.close() + self._lease_conn = None + return False + if done.endswith(" 0"): + logger.warning("本进程的登记不见了（可能被当成没了），重新登记") + await workers.register() + fence.renewed() + return True + + async def _lease_loop(self) -> None: + while True: + await self.renew() + await asyncio.sleep(workers.heartbeat_s()) + + def _on_fenced(self) -> None: + """隔离了：手上的执行由各处自己停（runner、job_runtime 登记的处理），这里换身份重新登记。""" + asyncio.ensure_future(self._reidentify()) + + async def _reidentify(self) -> None: + from app.cluster import fence + + old = identity.owner() + new = identity.regenerate() + leader.scheduler.until = None + try: + await workers.register() + fence.renewed() + logger.warning("隔离之后换了身份重新登记：%s → %s（之前认领的行由当班进程按「属主不在了」接手）", + identity.describe(old), identity.describe(new)) + except Exception: # noqa: BLE001 + logger.exception("隔离之后重新登记失败，下一拍再试") + + # ---------------- 当班、心跳的其余活 ---------------- + async def _lead(self) -> bool: was = leader.scheduler.held() try: @@ -113,10 +198,10 @@ class _Service: return now_leading async def beat(self) -> None: - """心跳的一拍（测试里直接调）。""" - if not await workers.renew(): - logger.warning("本进程的登记不见了（可能被当成没了），重新登记") - await workers.register() + """心跳的一拍（续租约另在 _lease_loop；测试里直接调）。SQLite 只做本进程自己攒着的活。""" + if self.single: + await self._own_work() + return live = await workers.liveness() state.remember(live) if self.others - live.others: @@ -132,14 +217,17 @@ class _Service: await caches.poll() except Exception: # noqa: BLE001 logger.exception("读别的进程的缓存变化时出错") + await self._own_work() + if await self._lead() and (self.reap_now or time.monotonic() - self.last_reap >= workers.lease_s()): + # 隔一会儿的这一次信得过自己：属主是本进程的行（刚建、还没交给后台任务的）不当成没人管 + await self.reap(live.trusting()) + + async def _own_work(self) -> None: for name, fn in list(_beats): try: await fn() except Exception: # noqa: BLE001 logger.exception("心跳里的「%s」出错，下一拍再试", name) - if await self._lead() and (self.reap_now or time.monotonic() - self.last_reap >= workers.lease_s()): - # 隔一会儿的这一次信得过自己：属主是本进程的行（刚建、还没交给后台任务的）不当成没人管 - await self.reap(live.trusting()) async def reap(self, live: workers.Liveness | None = None) -> dict[str, Any]: self.reap_now = False @@ -173,20 +261,33 @@ class _Service: except Exception: # noqa: BLE001 - 一拍出错（库一时不通）不该让心跳停下 logger.exception("心跳这一拍出错，下一拍照常进行") + async def _light_loop(self) -> None: + while True: + await asyncio.sleep(workers.heartbeat_s()) + await self._own_work() + async def stop(self) -> None: """关停：停心跳、交班、删登记（别的进程马上接手本进程留下的行），停收发通知。""" if not self.started: return self.started = False - if self.task is not None: - self.task.cancel() - with contextlib.suppress(asyncio.CancelledError, Exception): - await self.task - self.task = None + from app.cluster import fence + + for task in (self.lease_task, self.task): + if task is not None: + task.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await task + self.task = self.lease_task = None + fence.deactivate() with contextlib.suppress(Exception): await leader.scheduler.release() with contextlib.suppress(Exception): await workers.unregister() + if self._lease_conn is not None: + with contextlib.suppress(Exception): + await self._lease_conn.close() + self._lease_conn = None signals.publish("gone", identity.owner()) await signals.stop() state.forget() diff --git a/backend/app/cluster/workers.py b/backend/app/cluster/workers.py index ca7edf45..adbd6ef2 100644 --- a/backend/app/cluster/workers.py +++ b/backend/app/cluster/workers.py @@ -2,7 +2,7 @@ - register / renew / unregister：启动时登记、隔一拍心跳续租约、平稳关停时删掉自己那一行（别的进程马上就知道它不在了， 不必等租约到期）。 -- liveness()：现在哪些别的进程还活着（租约没过期）。各种「启动整理」「接手」都只碰属主不在了的行： +- liveness()：现在哪些别的进程还活着（租约过期不到一个租约时长：再晚才算不在了，见 cluster/fence.py）。各种「启动整理」「接手」都只碰属主不在了的行： Liveness.alive(owner, local=…) —— 属主是别的进程：看它的租约；属主是自己：看这一行在本进程里是不是真的在跑（local）， 同一个进程里前一次起停留下的（测试里一个进程起停多次服务）照样算不在了；没有属主（升级前的行）算不在了。 @@ -65,9 +65,12 @@ async def liveness(now: datetime | None = None) -> Liveness: now = now or utcnow() me = identity.owner() + # 租约过期之后再多等一个租约时长才算不在了：卡住了但没死的进程（SIGSTOP、虚拟机暂停、库一时连不上）自己在一个租约时长时就 + # 停手（cluster/fence.py），两边中间留着一段，不会它还在写、这边已经接手 + floor = now - timedelta(seconds=lease_s()) async with SessionLocal() as session: ids = (await session.execute(select(ClusterWorker.id).where( - ClusterWorker.lease_until > now, ClusterWorker.id != me))).scalars().all() + ClusterWorker.lease_until > floor, ClusterWorker.id != me))).scalars().all() return Liveness(frozenset(ids)) diff --git a/backend/app/core/job_runtime.py b/backend/app/core/job_runtime.py index e860e0d7..a9f103e6 100644 --- a/backend/app/core/job_runtime.py +++ b/backend/app/core/job_runtime.py @@ -78,6 +78,11 @@ class SerialWriter: self._lock = asyncio.Lock() async def __call__(self, fn: Callable[..., Awaitable[Any]], *args: Any) -> Any: + from app.cluster import fence + + # 本进程续不上租约了（cluster/fence.py）：这个任务可能已经被别的进程接手，不再写 + fence.check() + async def go() -> Any: async with self._lock: async with SessionLocal() as session: @@ -509,14 +514,22 @@ class JobRun: await self.writer.settle() async def run(self) -> None: + from app.cluster import fence + + fence.bind() try: try: await self._queue() await self._run() + except fence.Fenced: + # 本进程续不上租约了：停手、不写收尾（这个任务由接手的进程按「属主不在了」处理） + logger.error("%s %s：本进程续不上租约，停手、不写收尾", self.kind, self.job_id) except asyncio.CancelledError: current = asyncio.current_task() if current is not None: current.uncancel() + if fence.fenced_here(): + return await self._stopped() except JobFailed as e: await self._failed(str(e)) @@ -568,9 +581,18 @@ class JobManager: self._factory = factory self.label = label self.name = name - from app.cluster import requests + from app.cluster import fence, requests requests.handle("stop", name)(self._stop_requested) + fence.on_trip(self._stop_fenced) + + def _stop_fenced(self) -> None: + """本进程隔离了（cluster/fence.py）：停掉手上的任务，停的时候不写收尾。""" + for job_id in list(self._runs): + run = self._runs.get(job_id) + if run is not None and run.task is not None and not run.task.done(): + run.stop_reason = run.stop_reason or "fenced" + run.task.cancel() def model(self) -> Any: """任务表（停止时读属主用）：缺省取 factory 上的 model。""" diff --git a/backend/app/db/base.py b/backend/app/db/base.py index 1747e6e1..b839df41 100644 --- a/backend/app/db/base.py +++ b/backend/app/db/base.py @@ -571,6 +571,7 @@ def _migrate(conn) -> None: # pragma: no cover - 同步回调 cols = columns.get(table) or set() if all(c in cols for c in wanted): conn.execute(text(f"CREATE INDEX IF NOT EXISTS {name} ON {table} ({', '.join(wanted)})")) + _unique_run_event_seq(conn, columns) for name, table, wanted, where, dedupe in _UNIQUE_INDEX_MIGRATIONS: cols = columns.get(table) or set() if not all(c in cols for c in wanted): @@ -583,6 +584,29 @@ def _migrate(conn) -> None: # pragma: no cover - 同步回调 _backfill_not_null(conn, columns) +def _unique_run_event_seq(conn: Any, columns: dict[str, set[str]]) -> None: # pragma: no cover - 同步回调 + """run_events 的 (run_id, seq) 补唯一索引（ux_run_events_run_seq，换掉原来不唯一的 ix_run_events_run_seq）。存量库里可能已经 + 有撞了号的（多进程之前的进程内计数、按需裁判的老写法）：先数一数，有就照实记日志、这一次不建（照旧用不唯一的索引），不让启动 + 失败——那几次运行的封存核对本来就对不上，删哪一条都不对，要人看。""" + from sqlalchemy import inspect, text + + if "run_events" not in columns: + return + names = {ix["name"] for ix in inspect(conn).get_indexes("run_events")} + if "ux_run_events_run_seq" in names: + return + dupes = conn.execute(text("SELECT run_id, seq, COUNT(*) FROM run_events GROUP BY run_id, seq HAVING COUNT(*) > 1 " + "LIMIT 20")).all() + if dupes: + runs = sorted({str(r[0]) for r in dupes}) + logger.warning("run_events 里有撞了号的事件（%d 处以上，涉及运行 %s）：这一次不建 (run_id, seq) 的唯一索引。" + "这几次运行的封存核对对不上，请人工核对后处理", len(dupes), "、".join(runs[:5])) + return + conn.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ux_run_events_run_seq ON run_events (run_id, seq)")) + if "ix_run_events_run_seq" in names: + conn.execute(text("DROP INDEX IF EXISTS ix_run_events_run_seq")) + + #: init_db 在 PostgreSQL 上拿的会话级咨询锁的号：几个进程同时启动时，建表和迁移一个一个来（同一个号别处别用） _INIT_LOCK = 0x4167656E744C6162 # "AgentLab" diff --git a/backend/app/db/models.py b/backend/app/db/models.py index b010432a..bf7f41ce 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -721,7 +721,9 @@ class RunEvent(Base): """完整事件流落库 —— 这既是 trace，也是页面刷新后重建时间线的依据。""" __tablename__ = "run_events" - __table_args__ = (Index("ix_run_events_run_seq", "run_id", "seq"),) + # (run_id, seq) 唯一（编排完善第 2 波 W2-b）：撞号时报错，不悄悄写两条——封存按 seq 排序算哈希，撞了号核对时对时不对。 + # 存量库由 db/base._migrate 补建（先查有没有重复，有就照实记日志、不建） + __table_args__ = (Index("ux_run_events_run_seq", "run_id", "seq", unique=True),) id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) run_id: Mapped[str] = mapped_column(ForeignKey("runs.id", ondelete="CASCADE")) diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 552af462..6599ca5b 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -15,7 +15,8 @@ from langgraph.errors import GraphRecursionError from langgraph.types import Command from sqlalchemy import case, func, select, update -from app.cluster import identity +from app.cluster import fence, identity +from app.cluster.fence import Fenced from app.core.bus import bus from app.core.config import settings from app.core.errors import describe_exception, raw_detail @@ -92,7 +93,7 @@ class RunManager: from app.engine.checkpoints import open_saver self._cm = open_saver() - self._checkpointer = await self._cm.__aenter__() + self._checkpointer = _fenced_saver(await self._cm.__aenter__()) # 上次进程被强杀时留下的记录。running 和 queued 要分开处理： # running 至少跑过一步、有 checkpoint，可以续；queued 还在排队， # 一个 checkpoint 都没写过，"恢复"它只会从 START 用默认值重跑一遍， @@ -209,6 +210,8 @@ class RunManager: persist: bool = True, ts: float | None = None, ) -> None: + # 执行这次运行的那段（cluster/fence 绑定过世代的）写之前问一句：租约续晚了就不写（可能已经被别的进程接手） + fence.check() event = self._event(run_id, event_type, node_id=node_id, data=data, ts=ts) # 先落库，再广播。顺序反过来会漏事件，而且是结构性的漏： # @@ -791,6 +794,11 @@ class RunManager: # 等待从 run.interrupted 算起——以前执行一直计到 finally，收尾那一截执行和等待两边都算， # active + wait 比墙钟还长（机器一忙就差出几十毫秒） paused: float | None = None + # 这段执行属于现在这一世代（cluster/fence.py）：本进程隔离之后，它不再写 + fence.bind() + mine = identity.owner() + #: 被隔离了（续不上租约）：停手，不写终态（这次运行由接手的进程按「属主不在了」处理） + fenced = False # 先要并发名额（app/cluster/slots.py）。等名额的时候被停止（或者关停）也要照实收尾，见 _stopped_in_queue slot = slots.hold(run_id, origin_kind) try: @@ -806,7 +814,7 @@ class RunManager: row = await session.get(Run, run_id) if row is not None: row.status = "running" - row.owner = identity.owner() + row.owner = mine # started_at 只在第一次开始时写。以前每段 _drive 都覆盖它， # 审批恢复、接着跑之后，运行的"开始时间"就成了最后一段的开始 if row.started_at is None: @@ -891,7 +899,13 @@ class RunManager: elif snapshot.next: status = "interrupted" + except Fenced: + fenced = True + logger.error("运行 %s：本进程续不上租约，停手、不写终态（由接手的进程按「属主不在了」处理）", run_id) except asyncio.CancelledError: + if fence.fenced_here(): + fenced = True + raise if self._closing: # 服务关停（dev.sh 的 --reload 改一个文件就是一次）不是用户取消。 # checkpoint 完好，记成中断、重启后从断点接着跑——和强杀后启动 @@ -956,11 +970,12 @@ class RunManager: elapsed = int((stop - started) * 1000) self._finalizing.add(run_id) try: - await self._finalize(run_id, status, error, final_state, elapsed, - terminal=terminal, error_node=error_node, end_reason=end_reason) + if not fenced: + await self._finalize(run_id, status, error, final_state, elapsed, terminal=terminal, + error_node=error_node, end_reason=end_reason, owner=mine) # 记 action.unknown、安排核实也算收尾：做完才把自己摘掉。以前摘掉之后才做，等运行「不在跑了」 # 再去读事件的一方（接口、测试）有一小段读不到这一条；库慢一点（PostgreSQL 满载时）就读不到 - if pending_action is not None: + if pending_action is not None and not fenced: from app.process.actions import on_unknown try: @@ -978,7 +993,7 @@ class RunManager: # 不收的话每跑一次带代码节点的图就多一台常驻 microVM（几百 MB）， # 而且没有任何自动回收路径会碰它——闲置回收只在"下次有人执行 # 代码"时才触发。interrupted 要留着，人回来还得接着跑。 - if status in ("succeeded", "failed", "cancelled"): + if status in ("succeeded", "failed", "cancelled") and not fenced: with contextlib.suppress(Exception): from app.sandbox.manager import sandbox_manager @@ -992,11 +1007,13 @@ class RunManager: try: if self._closing: return + if fence.fenced_here(): + 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) + end_reason=END_STOPPED, owner=identity.owner()) finally: self._finalizing.discard(run_id) self._tasks.pop(run_id, None) @@ -1021,6 +1038,7 @@ class RunManager: interrupts = chunk.get("__interrupt__") if not interrupts: return False + fence.check() for item in interrupts: value = getattr(item, "value", item) @@ -1082,7 +1100,9 @@ class RunManager: terminal: tuple[EventType, dict[str, Any]] | None = None, error_node: str | None = None, end_reason: str | None = None, + owner: str | None = None, ) -> None: + """收尾。owner：执行这一段的属主——运行那一行的属主已经不是它（被别的进程当成没了接手了，cluster/fence.py）就什么都不写。""" usage = dict(state.get("usage") or {}) output = dict(state.get("output") or {}) now = datetime.now(timezone.utc) @@ -1100,7 +1120,13 @@ class RunManager: run_class: str | None = None origin_kind: str | None = None async with SessionLocal() as session: - run = await session.get(Run, run_id) + # 先占住这一行再看属主（PostgreSQL 上 FOR UPDATE；SQLite 只有一个进程）：接手的进程改属主用的是带条件的 UPDATE， + # 两边一个接一个，被接手了的这边不写 + run = (await session.execute(select(Run).where(Run.id == run_id).with_for_update())).scalar_one_or_none() + if run is not None and owner is not None and run.owner != owner: + logger.error("运行 %s 已经被别的进程接手（属主 %s），本进程不写终态", run_id, identity.describe(run.owner)) + await session.rollback() + return if run: app_id = run.app_id run_class, origin_kind = run.run_class, run.origin_kind @@ -1759,6 +1785,31 @@ async def _timing(session: Any, run: Run, elapsed_ms: int, now: datetime) -> dic run_manager = RunManager() +def _fenced_saver(saver: BaseCheckpointSaver) -> BaseCheckpointSaver: + """断点库的写也要过 cluster/fence：本进程续不上租约之后，手上的执行不再写断点（接手的进程会从断点接着跑，两边都写就分叉了）。""" + for name in ("aput", "aput_writes"): + original = getattr(saver, name, None) + if original is None: + continue + + def guarded(*args: Any, __original: Any = original, **kwargs: Any) -> Any: + fence.check() + return __original(*args, **kwargs) + + setattr(saver, name, guarded) + return saver + + +def _stop_fenced() -> None: + """隔离了：停掉本进程在执行的运行（停的时候不写终态，见 _drive 的 fenced）。""" + for task in list(run_manager._tasks.values()): + if not task.done(): + task.cancel() + + +fence.on_trip(_stop_fenced) + + def _register_requests() -> None: """别的进程发来的：停止本进程在执行的运行、往它里面记一条事件（app/cluster/requests.py）。""" from app.cluster import requests diff --git a/backend/app/process/actions.py b/backend/app/process/actions.py index 6231d047..2b6a6393 100644 --- a/backend/app/process/actions.py +++ b/backend/app/process/actions.py @@ -399,6 +399,10 @@ async def send(prep: Prepared, capability: Any, *, run_id: str, node_id: str, ex else "演练：探索运行里端点没有测试环境，没有发送请求") return _output(action_id, prep, "simulated", None, receipt) + # 本进程续不上租约了（cluster/fence.py）就不发：这次运行可能已经被别的进程接手 + from app.cluster import fence + + fence.check() # 先记 sent 再发：发到一半进程没了，重启后看到的是 sent，只能核实，不会当成没发过再发一次 if not await _set(action_id, ("pending",), state="sent", request_at=utcnow()): raise ActionPending(action_id) diff --git a/backend/tests/test_cluster_fence.py b/backend/tests/test_cluster_fence.py new file mode 100644 index 00000000..902e81fe --- /dev/null +++ b/backend/tests/test_cluster_fence.py @@ -0,0 +1,149 @@ +"""自我隔离（cluster/fence.py，复核 R-b1）：本进程卡住了、续不上租约，别的进程已经把它在跑的运行接手了——它缓过来之后不再写： +不写事件（不和接手的一边撞号）、不写断点、不写终态（不把接手的结果盖掉）。在一个进程里装出来：把「上次续上租约」拨到一个租约 +时长以前，再把运行改成接手之后的样子。两种库都跑。""" +from __future__ import annotations + +import asyncio +import time + +import pytest +from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError + +from app.cluster import fence +from app.cluster.workers import lease_s +from app.db.base import SessionLocal, new_id +from app.db.models import Run, RunEvent +from app.engine import compiler +from app.engine.runner import run_manager +from app.engine.schema import NodeType + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + fence.activate() + yield + fence.deactivate() + await run_manager.shutdown() + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +GRAPH = {"nodes": [node("start", "input", fields=[{"name": "question"}]), + node("slow", "transform", mode="template", template="慢的这步"), + node("next", "transform", mode="template", template="后面这步"), + node("out", "output", fields=[{"name": "r", "value": "{{ nodes.next }}"}])], + "edges": [{"source": "start", "target": "slow"}, {"source": "slow", "target": "next"}, + {"source": "next", "target": "out"}]} + + +@pytest.fixture +def slow(monkeypatch): + """slow 那一步：第一次跑时卡住，等测试放行（好在它卡着的时候装出「续不上租约、被接手」）；之后立刻返回。""" + calls = {"n": 0} + gate = asyncio.Event() + entered = asyncio.Event() + original = compiler.RUNNERS[NodeType.TRANSFORM] + + async def runner(state, ctx): + if ctx.node.id == "slow": + calls["n"] += 1 + if calls["n"] == 1: + entered.set() + await gate.wait() + return await original(state, ctx) + + monkeypatch.setitem(compiler.RUNNERS, NodeType.TRANSFORM, runner) + return {"calls": calls, "gate": gate, "entered": entered} + + +async def _taken_over(run_id: str) -> None: + """别的进程看它的租约过期了、接手：标中断、属主清空（同 run_manager.reap 做的）。""" + from sqlalchemy import update + + async with SessionLocal() as session: + await session.execute(update(Run).where(Run.id == run_id).values( + status="interrupted", owner=None, error="服务重启，运行已挂起，可从断点继续运行")) + await session.commit() + + +async def _events(run_id: str) -> list[RunEvent]: + async with SessionLocal() as session: + return list((await session.execute(select(RunEvent).where(RunEvent.run_id == run_id) + .order_by(RunEvent.seq))).scalars()) + + +async def _no_repeated_seq(run_id: str) -> None: + async with SessionLocal() as session: + rows = (await session.execute(select(RunEvent.seq, func.count()).where(RunEvent.run_id == run_id) + .group_by(RunEvent.seq).having(func.count() > 1))).all() + assert rows == [], f"撞号：{rows}" + + +async def _stall_and_take_over(slow) -> str: + run = await run_manager.start(graph=GRAPH, input_payload={"question": "q"}) + await asyncio.wait_for(slow["entered"].wait(), timeout=15) + before = fence.generation() + fence.renewed(time.monotonic() - lease_s() - 1) # 卡了一个多租约时长 + await _taken_over(run.id) + slow["gate"].set() # 缓过来，手上的执行接着往下走 + for _ in range(200): + if not run_manager.is_active(run.id): + break + await asyncio.sleep(0.05) + assert not run_manager.is_active(run.id), "执行停了" + assert fence.generation() == before + 1, "发现续晚了：隔离了一次" + return run.id + + +async def test_a_stalled_executor_does_not_overwrite_the_takeover(slow): + run_id = await _stall_and_take_over(slow) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + assert row.status == "interrupted" and row.owner is None, "接手之后的样子没被盖掉" + events = await _events(run_id) + assert not [e for e in events if e.type in ("run.finished", "run.failed", "run.cancelled")] + assert not [e for e in events if e.node_id == "next"], "缓过来之后一条都没写" + await _no_repeated_seq(run_id) + + +async def test_a_run_continued_elsewhere_finishes_once_and_the_stalled_side_stays_quiet(slow): + run_id = await _stall_and_take_over(slow) + fence.renewed() # 换了身份重新登记之后，这一世代照常 + await run_manager.continue_failed(run_id) # 接手的那一边接着跑 + for _ in range(300): + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in ("succeeded", "failed") and row.manifest_hash: + break + await asyncio.sleep(0.05) + assert row.status == "succeeded", row.error + events = await _events(run_id) + assert len([e for e in events if e.type == "run.finished"]) == 1 + assert len([e for e in events if e.type == "node.started" and e.node_id == "next"]) == 1, "后面那一步只跑了一次" + await _no_repeated_seq(run_id) + + +async def test_the_final_write_checks_the_owner(): + async with SessionLocal() as session: + run = Run(status="running", owner="peer-host:1:abcd", graph={}, input={}) + session.add(run) + await session.commit() + await run_manager._finalize(run.id, "succeeded", None, {}, 10, owner="mine:1:ffff") + async with SessionLocal() as session: + row = await session.get(Run, run.id) + assert row.status == "running" and row.finished_at is None, "属主已经不是自己：不写终态" + + +async def test_a_repeated_seq_is_refused_by_the_database(): + async with SessionLocal() as session: + run = Run(status="running", graph={}, input={}) + session.add(run) + await session.flush() + session.add_all([RunEvent(id=new_id(), run_id=run.id, seq=1, type="log", ts=0, data={}), + RunEvent(id=new_id(), run_id=run.id, seq=1, type="log", ts=0, data={})]) + with pytest.raises(IntegrityError): + await session.commit() diff --git a/backend/tests/test_multiprocess_e2e.py b/backend/tests/test_multiprocess_e2e.py index d66c7f16..7181b10e 100644 --- a/backend/tests/test_multiprocess_e2e.py +++ b/backend/tests/test_multiprocess_e2e.py @@ -11,7 +11,9 @@ 全站 2 个、给交互留底 1 个：评测借满了别的名额、还有题排着，问数据一来马上拿到留底，不等评测跑完； 6. 每日花费上限跨进程不越过：两个进程同时往同一本账上记，记下的合计不超过上限； 7. 强杀 B：租约过期后，B 上在跑的运行被照实标中断（可从断点继续），停在审批上的运行照常能在 A 上批、接着跑完；B 当着调度班的 - 话 A 接班。 + 话 A 接班； +8. 卡住但没死（复核 R-b1）：A 在跑一次运行时被 SIGSTOP，B 等它的租约过期两倍时长后接手、标中断；A 缓过来（SIGCONT）之后发现续晚了， + 停手、不写终态、不写事件，换一个身份重新登记——不管 B 有没有在这期间把那次运行接着跑完，都不双写、不盖掉终态、序号不重。 缺省不跑：要 AGENTLAB_MULTIPROCESS_E2E=1 和 PostgreSQL 测试库（scripts/test-multiprocess.sh 起临时容器、设好这两样）。 """ @@ -88,8 +90,16 @@ class Worker: self.proc.send_signal(signal.SIGKILL) self.proc.wait(timeout=10) + def pause(self) -> None: + self.proc.send_signal(signal.SIGSTOP) + + def resume(self) -> None: + if self.proc.poll() is None: + self.proc.send_signal(signal.SIGCONT) + def stop(self) -> None: if self.proc.poll() is None: + self.proc.send_signal(signal.SIGCONT) self.proc.send_signal(signal.SIGTERM) try: self.proc.wait(timeout=15) @@ -98,7 +108,7 @@ class Worker: def tail(self) -> str: try: - return self.log.read_text(encoding="utf-8", errors="replace")[-4000:] + return self.log.read_text(encoding="utf-8", errors="replace")[-12000:] except OSError: return "" @@ -420,7 +430,8 @@ async def test_5b_the_interactive_reserve_is_there_when_evaluations_fill_the_res put = await client.put(f"{a.base}/api/settings", json={"values": {"concurrency": { "max_total": 2, "interactive_reserve": 1, "classes": {}}}}) assert put.status_code == 200, put.text - assert put.json()["concurrency"]["classes"]["system"]["limit"] == 1, "别的类缺省借满到 全站 − 留底" + got = put.json()["concurrency"] + assert got["max_total"] - got["interactive_reserve"] == 1, "别的类最多借得到 全站 − 留底 = 1 个" slots.forget_policy() monkeypatch.setattr(settings, "sandbox_backend", "local") await cluster.start() @@ -480,6 +491,88 @@ async def _spent() -> float: return float((await Ledger("e2e_spend").daily())["usd"]) +async def _stall_a_during_a_run(client, a: Worker, b: Worker, seconds: float) -> str: + """A 上发起一次运行，跑起来之后 SIGSTOP A，等 B 接手（标中断）。交回运行 id（A 还停着，调用方负责 SIGCONT）。""" + run_id = (await client.post(f"{a.base}/api/runs", json={"graph": slow_graph(seconds), "input": {"question": "q"}, + "origin": "canvas"})).json()["id"] + await _until(lambda: _owner_while_active(run_id), what="A 认领了运行") + await asyncio.sleep(0.5) + a.pause() + + async def taken_over() -> Any: + row = await _run(client, b, run_id) + return row if row["status"] == "interrupted" else None + + await _until(taken_over, timeout=LEASE_S * 6 + 10, what="B 接手了卡住的 A 上的运行") + return run_id + + +async def _a_reidentified(client, a: Worker) -> None: + """A 缓过来之后换了身份重新登记：fixture 里记的 owner 跟着换。""" + old = a.owner + + async def fresh() -> Any: + try: + me = (await client.get(f"{a.base}/api/system")).json()["cluster"]["me"] + except Exception: # noqa: BLE001 + return None + return me if me != old else None + + a.owner = await _until(fresh, timeout=15, what="A 换身份重新登记") + + +async def _settled_events(client, worker: Worker, run_id: str) -> list[dict[str, Any]]: + return (await client.get(f"{worker.base}/api/runs/{run_id}/events")).json() + + +def _seqs_unique(events: list[dict[str, Any]]) -> None: + seqs = [e["seq"] for e in events] + assert len(seqs) == len(set(seqs)), f"撞号：{sorted(seqs)}" + + +async def test_6b_a_stalled_worker_does_not_write_after_it_was_taken_over(workers): + a, b = workers["A"], workers["B"] + async with await _client() as client: + try: + run_id = await _stall_a_during_a_run(client, a, b, 6) + finally: + a.resume() + await _a_reidentified(client, a) + await asyncio.sleep(max(4.0, 3 * HEARTBEAT_S)) # A 手上那一步本来再过几秒就跑完：给它机会（它不该写） + row = await _run(client, b, run_id) + assert row["status"] == "interrupted", f"接手之后的状态被盖掉了：{row['status']}" + events = await _settled_events(client, b, run_id) + assert not [e for e in events if e["type"] in ("run.finished", "run.failed", "run.cancelled")] + assert any(e["type"] == "log" and e["data"].get("code") == "server_shutdown" for e in events) + _seqs_unique(events) + + +async def test_6c_a_run_continued_by_the_other_worker_is_not_written_twice(workers): + a, b = workers["A"], workers["B"] + async with await _client() as client: + try: + run_id = await _stall_a_during_a_run(client, a, b, 3) + cont = await client.post(f"{b.base}/api/runs/{run_id}/continue", json={}) + assert cont.status_code == 200, cont.text + finally: + a.resume() + await _a_reidentified(client, a) + + async def done() -> Any: + row = await _run(client, b, run_id) + return row if row["status"] in ("succeeded", "failed") and row.get("manifest_hash") else None + + row = await _until(done, timeout=60, what="B 接着跑完") + await asyncio.sleep(2.0) + assert row["status"] == "succeeded", row + assert (await _run(client, b, run_id))["status"] == "succeeded", "A 缓过来之后没有把终态改掉" + events = await _settled_events(client, b, run_id) + assert len([e for e in events if e["type"] == "run.finished"]) == 1 + _seqs_unique(events) + verified = (await client.get(f"{b.base}/api/runs/{run_id}/verify")).json() + assert verified["ok"] is True, verified + + async def test_7_kill_a_worker(workers): a, b = workers["A"], workers["B"] async with await _client() as client: