diff --git a/backend/app/cluster/fence.py b/backend/app/cluster/fence.py index ad18d85f..1d94b3d6 100644 --- a/backend/app/cluster/fence.py +++ b/backend/app/cluster/fence.py @@ -6,9 +6,10 @@ - 续租约在单独的任务、单独的连接上做(cluster/service.py),不被同一拍别的活拖住;最近一次续上的时刻记在这里(renewed)。 - 本进程的执行在动手写之前问一句 check():离上次续上已经超过一个租约时长(别的进程按两个租约时长才接手,中间留着一段),或者这段 - 执行是在上一次隔离之前开始的(世代不对),就抛 Fenced——写事件、写断点、写终态、写待办和动作、发外部写入之前都问。 + 执行是在上一次隔离之前开始的(世代不对),就抛 Fenced——写事件、写断点、写终态、写待办和动作、发外部写入之前都问。只对绑定了 + 世代的执行(运行的一段执行、后台任务)抛;别的调用者问到时只隔离、不抛。 - 一旦发现续晚了(trip):世代加一,手上的执行全部停掉(停的时候不写终态,那些行交给接手的进程按「属主不在了」处理),换一个新的 - 身份重新登记,之后发起的照常。 + 身份重新登记;新身份登记上之前新开始的执行先等着(ready),登记上之后照常。 **用哪只钟(复核 R-b7)。** 别的进程判「租约过期」用的是库的时钟(cluster/workers:续租写 lease_until = 库的 now() + 租约), 跟各台机器的墙钟差多少无关。这边判「续晚了」用计入休眠的钟:time.monotonic() 在 macOS(mach_absolute_time)、Linux @@ -21,6 +22,7 @@ Fenced 是 BaseException:节点、工具里常见的 except Exception 拦不 """ from __future__ import annotations +import asyncio import contextvars import logging import sys @@ -96,17 +98,38 @@ def bind() -> None: def stale() -> bool: - """离上次续上租约是不是已经超过一个租约时长:两只钟(计入休眠的单调钟、墙钟)里走得多的那只说了算。""" + """离上次续上租约是不是已经超过一个租约时长:两只钟(计入休眠的单调钟、墙钟)里走得多的那只说了算。隔离之后、新身份还没 + 登记上的那一段也算(pending)。""" if not _active: return False since = _since() if since is None: - return False + return True from app.cluster.workers import lease_s return max(since) > lease_s() +def pending() -> bool: + """隔离过、换的新身份还没登记上。""" + return _active and _renewed is None + + +async def ready(timeout: float | None = None) -> None: + """新开始的一段执行(运行、后台任务)在 bind() 之前等一等:隔离之后、新身份还没登记上时,它认领的行在别的进程眼里属主 + 不在,会被当成没人管接手(复核 R-b9)。最多等一个租约时长(timeout),等不到抛 Fenced——不开始,那几行由别的进程照「属主 + 不在了」处理(cluster/service 那边同时会放弃这个一直没登记上的身份)。""" + if not pending(): + return + from app.cluster.workers import lease_s + + deadline = _clock() + (lease_s() if timeout is None else timeout) + while pending(): + if _clock() >= deadline: + raise Fenced() + await asyncio.sleep(0.05) + + def jumped() -> bool: """两只钟差出一个租约以上(墙钟往前或往后跳了,或者单调钟没计入休眠):续租时问,是就隔离。""" if not _active: @@ -120,22 +143,22 @@ def jumped() -> bool: def fenced_here() -> bool: - """当前这段执行还能不能写:世代不对(隔离之前开始的)或者租约续晚了。""" + """当前这段执行还能不能写:世代不对(隔离之前开始的)或者租约续晚了。只说绑定了世代的执行(bind 过的)。""" if not _active: return False bound = _bound.get() - return (bound is not None and bound != _generation) or stale() + return bound is not None and (bound != _generation or stale()) def check() -> None: - """写之前问一句:不许写就抛 Fenced(续晚了的话顺手隔离)。""" + """写之前问一句。续晚了就隔离(谁先问到都一样);抛 Fenced 只对绑定了世代的执行(运行的一段执行、后台任务:bind 过的)—— + 别的调用者(HTTP 处理、调度那一拍、进程间请求)不是认领了行在跑的执行,抛给它们只会把调度循环之类的打停(复核 R-b9)。""" if not _active: return - if stale(): + if _renewed is not None and stale(): trip("执行要写的时候发现租约已经续晚了") - raise Fenced() bound = _bound.get() - if bound is not None and bound != _generation: + if bound is not None and (bound != _generation or stale()): raise Fenced() @@ -162,4 +185,4 @@ def trip(why: str) -> bool: __all__ = ["Fenced", "SLEEP_AWARE", "activate", "bind", "check", "deactivate", "fenced_here", "generation", "jumped", "now", - "on_trip", "renewed", "stale", "trip"] + "on_trip", "pending", "ready", "renewed", "stale", "trip"] diff --git a/backend/app/cluster/requests.py b/backend/app/cluster/requests.py index 34fb7e8f..2711d4d6 100644 --- a/backend/app/cluster/requests.py +++ b/backend/app/cluster/requests.py @@ -94,9 +94,11 @@ _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: + from app.cluster.fence import Fenced + try: await fn(target_id, actor, payload) - except Exception: # noqa: BLE001 + except (Exception, Fenced): # noqa: BLE001 - Fenced 是 BaseException,不接住就成了没人取的任务异常 logger.exception("办进程间请求 %s/%s %s 出错", kind, target, target_id) diff --git a/backend/app/cluster/service.py b/backend/app/cluster/service.py index 1fbd7ace..2d97a18b 100644 --- a/backend/app/cluster/service.py +++ b/backend/app/cluster/service.py @@ -65,6 +65,11 @@ class _Service: #: SQLite:守卫已经确认只有本进程——登记只写一次、不续租、本进程直接当班、接手只在启动时做(不为协调多写库) self.single = False self._lease_conn: Any = None + self.lease_wake: asyncio.Event | None = None + #: 隔离之后、新身份还没登记上:从什么时候起(cluster/fence.now 的读数) + self.pending_since: float | None = None + #: 隔离掉(或者一直没登记上、放弃了)的旧身份:新身份登记上之后收拾它们留下的登记、名额、请求、当班租约 + self.retired: list[str] = [] async def start(self) -> None: """登记、开始收发通知、起心跳。SQLite 时先取单进程的守卫(拿不到抛 guard.SingleProcessOnly)。""" @@ -94,6 +99,7 @@ class _Service: await self._lead() fence.activate() fence.on_trip(self._on_fenced) + self.lease_wake = asyncio.Event() 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()), @@ -159,27 +165,62 @@ class _Service: return True async def _lease_loop(self) -> None: + """续租约;隔离之后改成重新登记新身份。哪一下出了错(库一抖、版本守卫拦下)都不让这个任务退出——它一退,之后就再也没有 + 隔离保护了(复核 R-b9):照实记日志,退避着重试(最长一个租约时长)。""" + from app.cluster import fence + + delay = workers.heartbeat_s() while True: - await self.renew() - await asyncio.sleep(workers.heartbeat_s()) + try: + if self.pending_since is not None: + await self.register_again() + else: + await self.renew() + delay = workers.heartbeat_s() + except asyncio.CancelledError: + raise + except (Exception, fence.Fenced): # noqa: BLE001 + delay = min(max(delay * 2, workers.heartbeat_s()), workers.lease_s()) + logger.exception("%s出错,%g 秒后再试", "隔离之后重新登记" if self.pending_since is not None else "续租约", delay) + if self.lease_wake is None: + await asyncio.sleep(delay) + continue + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(self.lease_wake.wait(), timeout=delay) + self.lease_wake.clear() def _on_fenced(self) -> None: - """隔离了:手上的执行由各处自己停(runner、job_runtime 登记的处理),这里换身份重新登记。""" - asyncio.ensure_future(self._reidentify()) - - async def _reidentify(self) -> None: + """隔离了:手上的执行由各处自己停(runner、job_runtime 登记的处理);这里马上换身份(之后新建的行记新身份),叫续租的任务 + 去登记它。""" from app.cluster import fence old = identity.owner() + self.retired.append(old) new = identity.regenerate() + self.pending_since = fence.now() leader.scheduler.drop() - try: - await workers.register() - fence.renewed() - logger.warning("隔离之后换了身份重新登记:%s → %s(之前认领的行由当班进程按「属主不在了」接手)", - identity.describe(old), identity.describe(new)) - except Exception: # noqa: BLE001 - logger.exception("隔离之后重新登记失败,下一拍再试") + logger.warning("隔离:换成新身份 %s(旧身份 %s),登记上之前新开始的执行先等着", new, old) + if self.lease_wake is not None: + self.lease_wake.set() + + async def register_again(self) -> None: + """隔离之后登记新身份。一个租约时长都没登记上(库连不上、版本守卫拦着)就放弃这个身份、再换一个:这期间等不到的执行已经 + 放弃开始(fence.ready),它们的行记着这个身份——它永远不登记,别的进程就照「属主不在了」处理那几行。""" + from app.cluster import fence + + assert self.pending_since is not None + if fence.now() - self.pending_since > workers.lease_s(): + gave_up = identity.owner() + self.retired.append(gave_up) + identity.regenerate() + self.pending_since = fence.now() + logger.error("新身份 %s 一个租约时长都没登记上,放弃它、换成 %s 再试", gave_up, identity.owner()) + await workers.register() + retired, self.retired = self.retired, [] + self.pending_since = None + fence.renewed() + logger.warning("隔离之后换了身份重新登记:%s → %s(之前认领的行由当班进程按「属主不在了」接手)", + "、".join(retired), identity.owner()) # ---------------- 当班、心跳的其余活 ---------------- @@ -203,8 +244,9 @@ class _Service: return now_leading async def beat(self) -> None: - """心跳的一拍(续租约另在 _lease_loop;测试里直接调)。SQLite 只做本进程自己攒着的活。""" - if self.single: + """心跳的一拍(续租约另在 _lease_loop;测试里直接调)。SQLite 只做本进程自己攒着的活;隔离之后新身份还没登记上时也只做 + 这些(不争当班、不接手)。""" + if self.single or self.pending_since is not None: await self._own_work() return live = await workers.liveness() @@ -254,6 +296,8 @@ class _Service: return out async def _loop(self) -> None: + from app.cluster import fence + assert self.wake is not None while True: with contextlib.suppress(asyncio.TimeoutError): @@ -263,7 +307,7 @@ class _Service: await self.beat() except asyncio.CancelledError: raise - except Exception: # noqa: BLE001 - 一拍出错(库一时不通)不该让心跳停下 + except (Exception, fence.Fenced): # noqa: BLE001 - 一拍出错(库一时不通)不该让心跳停下 logger.exception("心跳这一拍出错,下一拍照常进行") async def _light_loop(self) -> None: @@ -284,6 +328,7 @@ class _Service: with contextlib.suppress(asyncio.CancelledError, Exception): await task self.task = self.lease_task = None + self.pending_since, self.retired = None, [] fence.deactivate() with contextlib.suppress(Exception): await leader.scheduler.release() diff --git a/backend/app/core/job_runtime.py b/backend/app/core/job_runtime.py index b050ee83..15775ee7 100644 --- a/backend/app/core/job_runtime.py +++ b/backend/app/core/job_runtime.py @@ -543,6 +543,11 @@ class JobRun: async def run(self) -> None: from app.cluster import fence + try: + await fence.ready() # 隔离之后新身份还没登记上时先等一等(cluster/fence.ready) + except fence.Fenced: + logger.error("%s %s:本进程隔离之后一直没重新登记上,不开始(由别的进程按「属主不在了」处理)", self.kind, self.job_id) + return fence.bind() try: try: diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 141b9b41..5d4c4af7 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -794,7 +794,12 @@ class RunManager: # 等待从 run.interrupted 算起——以前执行一直计到 finally,收尾那一截执行和等待两边都算, # active + wait 比墙钟还长(机器一忙就差出几十毫秒) paused: float | None = None - # 这段执行属于现在这一世代(cluster/fence.py):本进程隔离之后,它不再写 + # 这段执行属于现在这一世代(cluster/fence.py):本进程隔离之后,它不再写。隔离之后新身份还没登记上时先等一等 + try: + await fence.ready() + except Fenced: + logger.error("运行 %s:本进程隔离之后一直没重新登记上,不开始执行(由别的进程按「属主不在了」处理)", run_id) + return fence.bind() mine = identity.owner() #: 被隔离了(续不上租约):停手,不写终态(这次运行由接手的进程按「属主不在了」处理) diff --git a/backend/app/scheduler/loop.py b/backend/app/scheduler/loop.py index 7a7fba52..6c2a9f14 100644 --- a/backend/app/scheduler/loop.py +++ b/backend/app/scheduler/loop.py @@ -201,6 +201,7 @@ class _Runner: async def _loop(self, interval: float, stopping: asyncio.Event) -> None: from app.cluster import leading + from app.cluster.fence import Fenced from app.scheduler import webhook self._poked = asyncio.Event() @@ -218,7 +219,7 @@ class _Runner: self.notify = asyncio.create_task(_notify(webhook), name="agentlab-scheduler-notify") except asyncio.CancelledError: raise - except Exception: # noqa: BLE001 - 这一拍出错不该让循环停下 + except (Exception, Fenced): # noqa: BLE001 - 这一拍出错不该让循环停下(Fenced 是 BaseException:复核 R-b9) logger.exception("定时任务调度这一拍出错,下一拍照常进行") await self._wait(stopping, interval) diff --git a/backend/tests/test_cluster_fence.py b/backend/tests/test_cluster_fence.py index 18f99fd1..5d3908bb 100644 --- a/backend/tests/test_cluster_fence.py +++ b/backend/tests/test_cluster_fence.py @@ -4,7 +4,7 @@ from __future__ import annotations import asyncio -import time +import contextlib import pytest from sqlalchemy import func, select @@ -168,3 +168,108 @@ async def test_a_repeated_seq_is_refused_by_the_database(): RunEvent(id=new_id(), run_id=run.id, seq=1, type="log", ts=0, data={})]) with pytest.raises(IntegrityError): await session.commit() + + +# ---------------- 复核 R-b9:谁会被打停、谁不会 ---------------- + + +async def test_only_executions_bound_to_a_generation_are_stopped_by_check(): + """续晚之后第一个问 check() 的不一定是运行本身:没绑定世代的调用者(HTTP 处理、调度那一拍)只隔离、不抛;绑定了的抛。 + 隔离之后新身份还没登记上,新开始的执行在 ready() 上等,等不到抛 Fenced;登记上了就放行。""" + gate = asyncio.Event() + seen: list[str] = [] + + async def bound_writer() -> None: + fence.bind() + await gate.wait() + try: + fence.check() + seen.append("wrote") + except fence.Fenced: + seen.append("fenced") + + task = asyncio.create_task(bound_writer()) + await asyncio.sleep(0) + before = fence.generation() + fence.renewed(ago=lease_s() + 1) + fence.check() # 没绑定世代:只隔离 + assert fence.generation() == before + 1 and fence.pending() + gate.set() + await task + assert seen == ["fenced"], "绑定在隔离之前的那一段执行:不许再写" + + with pytest.raises(fence.Fenced): + await fence.ready(timeout=0.1) # 新身份还没登记上:新开始的执行等不到就不开始 + waiting = asyncio.create_task(fence.ready(timeout=5)) + await asyncio.sleep(0.1) + assert not waiting.done() + fence.renewed() # 登记上了 + await asyncio.wait_for(waiting, timeout=2) + fence.bind() + fence.check() # 新世代照常 + + +async def test_the_scheduler_loop_survives_a_fenced_tick(monkeypatch): + from app.scheduler import loop as sloop + + calls = {"n": 0} + + async def fire_due(now=None): + calls["n"] += 1 + if calls["n"] == 1: + raise fence.Fenced() + return [] + + monkeypatch.setattr(sloop, "fire_due", fire_due) + await sloop.runner.start(interval=0.02) + try: + for _ in range(200): + if calls["n"] >= 3: + break + await asyncio.sleep(0.02) + assert calls["n"] >= 3 and sloop.runner.running(), "被 Fenced 打到的那一拍之后照常往下" + finally: + await sloop.runner.stop() + + +async def test_the_lease_task_keeps_going_when_registering_again_fails(monkeypatch): + """隔离之后重新登记:库一抖、版本守卫拦下都不让续租的任务退出,退避着重试;登记上了照常续租。""" + from app.cluster import identity, service as service_mod, workers + from app.cluster.version import VersionMismatch + + monkeypatch.setattr(identity, "OWNER", identity.OWNER) # 测试完换回原来的身份 + monkeypatch.setattr(workers, "heartbeat_s", lambda: 0.01) + monkeypatch.setattr(workers, "lease_s", lambda: 5.0) + svc = service_mod._Service() + tries = {"register": 0, "renew": 0} + + async def register(*, exclusive: bool = False) -> None: # noqa: ARG001 + tries["register"] += 1 + if tries["register"] == 1: + raise VersionMismatch("库里还有版本不同的进程活着") + if tries["register"] == 2: + raise OSError("库一时连不上") + + async def renew() -> bool: + tries["renew"] += 1 + return True + + monkeypatch.setattr(workers, "register", register) + monkeypatch.setattr(svc, "renew", renew) + old = identity.owner() + fence.trip("测试") + svc._on_fenced() + assert identity.owner() != old and svc.pending_since is not None + task = asyncio.create_task(svc._lease_loop()) + try: + for _ in range(300): + if tries["renew"] >= 2: + break + await asyncio.sleep(0.01) + assert not task.done(), "续租的任务还在" + assert tries["register"] == 3 and svc.pending_since is None and not fence.pending() + assert tries["renew"] >= 2, "登记上之后照常续租" + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task