diff --git a/README.md b/README.md index 2b015bcc..e183bce6 100644 --- a/README.md +++ b/README.md @@ -1188,7 +1188,7 @@ asyncpg 方言一律这样做，同一个事务里、在同一条服务端连接 SQL 形状。各进程各自的并发闸门（口径比对、自动对比、起草的模型调用、上传解析）按进程算，进程多了总量跟着多。 **多台机器。** 几台机器要用同一份数据目录：放在共享存储（NFS 之类）上，每台挂在**同一个绝对路径**（库里记着上传表格各版本库文件的 -绝对路径），各台机器的时钟要对时（NTP；租约按墙钟算）。文件都是先写临时文件再改名，读的一方看不到写了一半的；写上传版本库、记录表 +绝对路径），各台机器的时钟仍要对时（NTP）：租约按库的时钟算，机器之间差多少都不影响谁算活着，但「停了两个租约时长」这类判断看的是各机器按本机时钟写下的时间。文件都是先写临时文件再改名，读的一方看不到写了一半的；写上传版本库、记录表 在库里的锁下一个接一个。但要说实话：数据目录里还有 SQLite 文件——上传表格的各版本库（发布后只读）、填报的记录表（`records/*.db`， 原地追加写）、沙箱里代码自己建的库——SQLite 文件放在网络文件系统上，读写一致性靠文件锁，而 NFS 上的文件锁不可靠：一台机器在追加时 另一台读，可能读到不一致的页。所以多台机器时，记录表写得多的部署建议用一台机器多进程，或者把数据目录放在支持可靠文件锁的存储上。 diff --git a/backend/app/cluster/fence.py b/backend/app/cluster/fence.py index 60da513b..ad18d85f 100644 --- a/backend/app/cluster/fence.py +++ b/backend/app/cluster/fence.py @@ -10,12 +10,20 @@ - 一旦发现续晚了（trip）：世代加一，手上的执行全部停掉（停的时候不写终态，那些行交给接手的进程按「属主不在了」处理），换一个新的 身份重新登记，之后发起的照常。 +**用哪只钟（复核 R-b7）。** 别的进程判「租约过期」用的是库的时钟（cluster/workers：续租写 lease_until = 库的 now() + 租约）， +跟各台机器的墙钟差多少无关。这边判「续晚了」用计入休眠的钟：time.monotonic() 在 macOS（mach_absolute_time）、Linux +（CLOCK_MONOTONIC）上都不计休眠——笔记本合盖、虚拟机挂起超过两个租约，对面早已接手，这边还以为刚续过。所以 Linux 用 +CLOCK_BOOTTIME，macOS 用 clock_gettime(CLOCK_MONOTONIC)（这个计入休眠），都取不到时退回 monotonic；不论哪种，另外同时比墙钟， +两只钟里走得多的那只说过了一个租约就算续晚了。续租时再比一下两只钟（jumped）：差出一个租约以上（墙钟跳了、或者计入休眠的钟 +其实没计）就隔离。续租那一句本身也按库的时钟核对：库里自己的租约已经过期（或者登记不见了），就不再续这个身份。 + Fenced 是 BaseException：节点、工具里常见的 except Exception 拦不住它，一路退到执行的最外层。没起协调（测试、脚本）时什么都不查。 """ from __future__ import annotations import contextvars import logging +import sys import time from collections.abc import Callable from typing import Any @@ -27,29 +35,55 @@ class Fenced(BaseException): """本进程续不上租约了（或者这段执行属于上一次隔离之前）：不许再写。""" +def _sleep_aware_clock() -> tuple[Callable[[], float], bool]: + """计入休眠的单调钟，和它是不是真的计入休眠。""" + if sys.platform.startswith("linux") and hasattr(time, "CLOCK_BOOTTIME"): + return (lambda: time.clock_gettime(time.CLOCK_BOOTTIME)), True + if sys.platform == "darwin" and hasattr(time, "CLOCK_MONOTONIC"): + return (lambda: time.clock_gettime(time.CLOCK_MONOTONIC)), True + return time.monotonic, False + + +_clock, SLEEP_AWARE = _sleep_aware_clock() +#: 墙钟（单独一个名字，测试里拨它） +_wall: Callable[[], float] = time.time + _generation = 0 _renewed: float | None = None +_renewed_wall: 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: +def now() -> float: + """计入休眠的单调钟的读数（本进程里判时长都用它：cluster/leader 判还当不当班也是）。""" + return _clock() + + +def activate() -> None: """起协调、续上了第一次租约时（PostgreSQL；SQLite 只有一个进程，不用隔离）。""" - global _active, _renewed + global _active _active = True - _renewed = time.monotonic() if renewed_at is None else renewed_at + renewed() def deactivate() -> None: - global _active, _renewed - _active, _renewed = False, None + global _active, _renewed, _renewed_wall + _active, _renewed, _renewed_wall = False, None, None -def renewed(at: float | None = None) -> None: - """续上了一次租约。""" - global _renewed - _renewed = time.monotonic() if at is None else at +def renewed(*, ago: float = 0.0) -> None: + """续上了一次租约（ago：其实是多少秒以前续上的，测试里装「卡了一阵」用）。""" + global _renewed, _renewed_wall + _renewed, _renewed_wall = _clock() - ago, _wall() - ago + + +def _since() -> tuple[float, float] | None: + """离上次续上：(计入休眠的钟走了多少, 墙钟走了多少)。""" + if _renewed is None or _renewed_wall is None: + return None + return _clock() - _renewed, _wall() - _renewed_wall def generation() -> int: @@ -61,13 +95,28 @@ def bind() -> None: _bound.set(_generation) -def stale(now: float | None = None) -> bool: - """离上次续上租约是不是已经超过一个租约时长。""" - if not _active or _renewed is None: +def stale() -> bool: + """离上次续上租约是不是已经超过一个租约时长：两只钟（计入休眠的单调钟、墙钟）里走得多的那只说了算。""" + if not _active: + return False + since = _since() + if since is None: return False from app.cluster.workers import lease_s - return (time.monotonic() if now is None else now) - _renewed > lease_s() + return max(since) > lease_s() + + +def jumped() -> bool: + """两只钟差出一个租约以上（墙钟往前或往后跳了，或者单调钟没计入休眠）：续租时问，是就隔离。""" + if not _active: + return False + since = _since() + if since is None: + return False + from app.cluster.workers import lease_s + + return abs(since[0] - since[1]) > lease_s() def fenced_here() -> bool: @@ -97,11 +146,11 @@ def on_trip(handler: Callable[[], Any]) -> None: def trip(why: str) -> bool: """隔离：世代加一，叫各处停掉手上的执行（cluster/service 再换身份重新登记）。已经在这一世代隔离过的不重复。交回这一次有没有动手。""" - global _generation, _renewed + global _generation, _renewed, _renewed_wall if not _active or _renewed is None: return False _generation += 1 - _renewed = None + _renewed = _renewed_wall = None logger.error("本进程的租约续晚了（%s）：当作认领的运行、问数据、后台任务可能已被别的进程接手，停掉手上的执行、不写终态，" "换一个身份重新登记", why) for handler in list(_handlers): @@ -112,5 +161,5 @@ def trip(why: str) -> bool: return True -__all__ = ["Fenced", "activate", "bind", "check", "deactivate", "fenced_here", "generation", "on_trip", "renewed", - "stale", "trip"] +__all__ = ["Fenced", "SLEEP_AWARE", "activate", "bind", "check", "deactivate", "fenced_here", "generation", "jumped", "now", + "on_trip", "renewed", "stale", "trip"] diff --git a/backend/app/cluster/leader.py b/backend/app/cluster/leader.py index fe169232..b532b11e 100644 --- a/backend/app/cluster/leader.py +++ b/backend/app/cluster/leader.py @@ -7,6 +7,9 @@ 两个进程短暂地都以为自己当班（续租那一拍库慢了、本地还没过期）时，重复不了：定时认领本来就是条件 UPDATE、触发记录 (任务, 时刻, 种类) 唯一，事件派发按投递记录的唯一约束，待办的升级、核实各自也是条件更新。当班只是让它们平常不撞。 本地另留一截余量（MARGIN）：租约快到期还没续上时，自己先当作不当班了。 + +租约按库的时钟写、按库的时钟判（复核 R-b7，同 cluster/workers）：机器之间墙钟差多少都不会让两边都以为租约归自己。本地判「还当 +不当班」不看墙钟，看计入休眠的单调钟（cluster/fence.now）：从发出续租那一刻起算一个租约时长，库里的租约只会更晚到期。 """ from __future__ import annotations @@ -15,9 +18,9 @@ from datetime import datetime, timedelta from sqlalchemy import case, or_, update -from app.cluster import identity -from app.cluster.workers import lease_s -from app.db.base import SessionLocal, utcnow +from app.cluster import fence, identity +from app.cluster.workers import db_now, lease_s +from app.db.base import SessionLocal logger = logging.getLogger(__name__) @@ -29,33 +32,42 @@ MARGIN_S = 2.0 class Leadership: def __init__(self, name: str) -> None: self.name = name + #: 库里记的到期时刻（库的时钟；给人看、交班用） self.until: datetime | None = None + #: 本地判还当不当班用的到期时刻（cluster/fence.now 的读数） + self.local_until: float | None = None self.epoch = 0 def single(self) -> None: """只有本进程（SQLite）：直接算当班，不写库。""" self.until = datetime.max.replace(tzinfo=None) + self.local_until = float("inf") - def held(self, now: datetime | None = None) -> bool: + def drop(self) -> None: + """本地当作不当班了（隔离之后换身份时：库里那份租约属于旧身份）。""" + self.until = self.local_until = None + + def held(self) -> bool: """本地看：还在当班（租约离到期还有余量）。""" - if self.until is None: + if self.until is None or self.local_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 + return fence.now() + margin < self.local_until - async def claim(self, now: datetime | None = None) -> bool: + async def claim(self) -> bool: """接班或续租。交回这一拍之后是不是当班。""" from app.db.models_cluster import ClusterLease from app.db.portable import insert_ignore - now = now or utcnow() - until = now + timedelta(seconds=lease_s()) + started = fence.now() me = identity.owner() mine = ClusterLease.owner == me async with SessionLocal() as session: + now = await db_now(session) + until = now + timedelta(seconds=lease_s()) done = await session.execute( update(ClusterLease).where( ClusterLease.name == self.name, @@ -74,15 +86,15 @@ class Leadership: .on_conflict_do_nothing(index_elements=["name"])) epoch = 1 if inserted.rowcount == 1 else None await session.commit() - was = self.held(now) + was = self.held() if epoch is None: if was or self.until is not None: logger.warning("调度当班交给了别的进程（本进程 %s 不再当班）", identity.describe(me)) - self.until = None + self.drop() return False if not was and self.epoch != epoch: logger.info("本进程（%s）接手调度当班（第 %d 班）", identity.describe(me), epoch) - self.until, self.epoch = until, int(epoch) + self.until, self.local_until, self.epoch = until, started + lease_s(), int(epoch) return True async def release(self) -> None: @@ -92,7 +104,7 @@ class Leadership: if self.until is None: return single = self.until.tzinfo is None - self.until = None + self.drop() if single: return async with SessionLocal() as session: diff --git a/backend/app/cluster/service.py b/backend/app/cluster/service.py index ff9cf7b3..1fbd7ace 100644 --- a/backend/app/cluster/service.py +++ b/backend/app/cluster/service.py @@ -123,23 +123,28 @@ class _Service: return self._lease_conn async def renew(self) -> bool: - """续一拍租约（在专用连接上，不走连接池：池子被占满时取连接要等，等得比租约还长）。续晚了（离上次续上超过一个租约时长） - 就不续这个身份了：隔离（cluster/fence.py），换身份重新登记。交回续没续上。""" - from datetime import timedelta - + """续一拍租约（在专用连接上，不走连接池：池子被占满时取连接要等，等得比租约还长）。按库的时钟续、按库的时钟核对（复核 + R-b7）：库里自己的租约已经过期（或者登记不见了），别的进程可能已经接手，不再续这个身份——隔离（cluster/fence.py），换身份 + 重新登记。本地的两只钟说续晚了、或者两只钟差出一个租约（墙钟跳了），同样隔离。交回续没续上。""" 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()) + if fence.jumped(): + fence.trip("续租约的时候发现本机的墙钟和单调钟差出一个租约以上（墙钟跳了，或者休眠没计进单调钟）") + return False 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()) + renewed = await asyncio.wait_for(conn.fetchval( + "UPDATE cluster_workers SET heartbeat_at = timezone('UTC', clock_timestamp()), " + "lease_until = timezone('UTC', clock_timestamp()) + make_interval(secs => $2) " + "WHERE id = $1 AND lease_until >= timezone('UTC', clock_timestamp()) RETURNING id", + identity.owner(), float(workers.lease_s())), timeout=workers.heartbeat_s()) + if renewed is None: + expired = await asyncio.wait_for(conn.fetchval( + "SELECT lease_until < timezone('UTC', clock_timestamp()) FROM cluster_workers WHERE id = $1", + identity.owner()), timeout=workers.heartbeat_s()) except Exception: # noqa: BLE001 - 库一时连不上：这一拍没续上，下一拍再试（续晚了的话下一拍隔离） logger.warning("续租约没续上，下一拍再试", exc_info=True) with contextlib.suppress(Exception): @@ -147,9 +152,9 @@ class _Service: await self._lease_conn.close() self._lease_conn = None return False - if done.endswith(" 0"): - logger.warning("本进程的登记不见了（可能被当成没了），重新登记") - await workers.register() + if renewed is None: + fence.trip("按库的时钟本进程的租约已经过期" if expired else "本进程的登记不见了（被当成不在了）") + return False fence.renewed() return True @@ -167,7 +172,7 @@ class _Service: old = identity.owner() new = identity.regenerate() - leader.scheduler.until = None + leader.scheduler.drop() try: await workers.register() fence.renewed() diff --git a/backend/app/cluster/version.py b/backend/app/cluster/version.py index 7020bd5a..c4450b6c 100644 --- a/backend/app/cluster/version.py +++ b/backend/app/cluster/version.py @@ -129,7 +129,6 @@ async def preflight(*, observe_s: float | None = None) -> None: if settings.db_dialect != "postgresql": return from app.cluster.workers import heartbeat_s, lease_s - from app.db.base import utcnow observe = 2 * heartbeat_s() + 1.0 if observe_s is None else observe_s loop = asyncio.get_running_loop() @@ -137,7 +136,7 @@ async def preflight(*, observe_s: float | None = None) -> None: first_beats: dict[str, Any] = {} warned = False while True: - mismatched = await _mismatched_alive(utcnow()) + mismatched = await _mismatched_alive() if not mismatched: return for peer in mismatched: @@ -154,14 +153,19 @@ async def preflight(*, observe_s: float | None = None) -> None: await asyncio.sleep(min(0.5, max(0.1, heartbeat_s() / 2))) -async def _mismatched_alive(now: Any) -> list[dict[str, Any]]: +async def _mismatched_alive(now: Any = None) -> list[dict[str, Any]]: + """租约没过期、版本不同的别的进程。now 不给就按库的时钟（租约是按库的时钟写的，cluster/workers）。""" + from sqlalchemy import text + from app.db.base import engine async with engine.connect() as conn: peers = await _peers(conn) + if peers and now is None: + now = (await conn.execute(text("SELECT timezone('UTC', clock_timestamp())"))).scalar() if not peers: return [] - naive = now.replace(tzinfo=None) + naive = _naive(now) alive = [p for p in peers if p.get("lease_until") is not None and _naive(p["lease_until"]) > naive] return [p for p in alive if not same(p.get("code_version"), p.get("schema_version"))] @@ -177,8 +181,9 @@ def _message(peers: list[dict[str, Any]]) -> str: def check_rows(rows: list[Any], now: Any) -> None: - """登记那一下（cluster/workers.register，持着事务级咨询锁）再查一遍：别的活着的进程里有版本不同的就抛 VersionMismatch。""" - naive = now.replace(tzinfo=None) + """登记那一下（cluster/workers.register，持着事务级咨询锁）再查一遍：别的活着的进程里有版本不同的就抛 VersionMismatch。 + now 是库的时钟。""" + naive = _naive(now) bad = [{"id": r.id, "code_version": r.code_version, "schema_version": r.schema_version} for r in rows if r.lease_until is not None and _naive(r.lease_until) > naive and r.id != identity.owner() and not same(r.code_version, r.schema_version)] diff --git a/backend/app/cluster/workers.py b/backend/app/cluster/workers.py index adbd6ef2..b71b0a2b 100644 --- a/backend/app/cluster/workers.py +++ b/backend/app/cluster/workers.py @@ -6,17 +6,19 @@ Liveness.alive(owner, local=…) —— 属主是别的进程：看它的租约；属主是自己：看这一行在本进程里是不是真的在跑（local）， 同一个进程里前一次起停留下的（测试里一个进程起停多次服务）照样算不在了；没有属主（升级前的行）算不在了。 -时间按应用的 utcnow（各台机器要对时，NTP）：租约以十秒计，几百毫秒的偏差无碍。 +租约用库的时钟（复核 R-b7）：PostgreSQL 上续租、登记写的是 lease_until = 库的 now() + 租约，判活比的也是库的 now()——租约由 +各台机器续、由别的机器判，用同一只钟，机器之间墙钟差多少都不起作用（cluster/leader 的当班租约同理）。SQLite 只有一个进程，用本机 +的钟。别处按「停了两个租约时长」判的时间（待办、动作的 updated_at）仍是各进程按本机时钟写的：各台机器还是要对时（NTP）。 """ from __future__ import annotations import logging import os from dataclasses import dataclass -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from typing import Any -from sqlalchemy import delete, select, update +from sqlalchemy import DateTime, delete, func, select, update from app.cluster import identity from app.core.config import settings @@ -38,6 +40,26 @@ def heartbeat_s() -> float: return float(settings.worker_heartbeat_s) +def _postgres(session: Any) -> bool: + return session.get_bind().dialect.name == "postgresql" + + +def clock(session: Any) -> Any: + """判租约用的「现在」（写进 UTCDateTime 列、跟它比的那种：UTC、不带时区）。PostgreSQL 是库的时钟（SQL 表达式），SQLite + 是本机的（只有一个进程）。""" + if _postgres(session): + return func.timezone("UTC", func.clock_timestamp(), type_=DateTime()) + return utcnow() + + +async def db_now(session: Any) -> datetime: + """库的时钟此刻的读数（带时区的 UTC）：要在 Python 里拿它算、比的时候用（登记、当班）。""" + if _postgres(session): + value = await session.scalar(select(func.timezone("UTC", func.clock_timestamp()))) + return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc) + return utcnow() + + @dataclass(frozen=True) class Liveness: """某一刻还活着的别的进程。""" @@ -60,15 +82,14 @@ class Liveness: async def liveness(now: datetime | None = None) -> Liveness: - """现在还活着（租约没过期）的别的进程。""" + """现在还活着（租约没过期）的别的进程。now 只给测试用（不给就是库的时钟）。""" from app.db.models_cluster import ClusterWorker - now = now or utcnow() me = identity.owner() # 租约过期之后再多等一个租约时长才算不在了：卡住了但没死的进程（SIGSTOP、虚拟机暂停、库一时连不上）自己在一个租约时长时就 # 停手（cluster/fence.py），两边中间留着一段，不会它还在写、这边已经接手 - floor = now - timedelta(seconds=lease_s()) async with SessionLocal() as session: + floor = (now if now is not None else clock(session)) - timedelta(seconds=lease_s()) ids = (await session.execute(select(ClusterWorker.id).where( ClusterWorker.lease_until > floor, ClusterWorker.id != me))).scalars().all() return Liveness(frozenset(ids)) @@ -93,8 +114,8 @@ async def register(*, exclusive: bool = False) -> None: from app.cluster import version me = identity.owner() - now = utcnow() async with SessionLocal() as session: + now = await db_now(session) if exclusive: await session.execute(delete(ClusterWorker).where(ClusterWorker.id != me)) elif session.get_bind().dialect.name == "postgresql": @@ -102,6 +123,7 @@ async def register(*, exclusive: bool = False) -> None: from sqlalchemy import text await session.execute(text("SELECT pg_advisory_xact_lock(:k)"), {"k": _REGISTER_KEY}) + now = await db_now(session) # 等锁可能等了一会儿：按拿到锁之后的库时钟 others = (await session.execute(select(ClusterWorker).where(ClusterWorker.id != me))).scalars().all() version.check_rows(list(others), now) row = await session.get(ClusterWorker, me) @@ -115,12 +137,14 @@ async def register(*, exclusive: bool = False) -> None: async def renew() -> bool: - """续一拍租约。自己那一行不见了（别的进程以为我没了、清掉了）返回 False，调用方重新登记。""" + """续一拍租约（走连接池的那一种；服务里续租在专用连接上，见 cluster/service.renew）。自己那一行不见了、或者按库的时钟已经 + 过期了返回 False（不续）。""" from app.db.models_cluster import ClusterWorker - now = utcnow() async with SessionLocal() as session: - done = await session.execute(update(ClusterWorker).where(ClusterWorker.id == identity.owner()).values( + now = clock(session) + done = await session.execute(update(ClusterWorker).where( + ClusterWorker.id == identity.owner(), ClusterWorker.lease_until >= now).values( heartbeat_at=now, lease_until=now + timedelta(seconds=lease_s())) .execution_options(synchronize_session=False)) await session.commit() @@ -139,8 +163,8 @@ async def prune(now: datetime | None = None) -> int: """删掉早就过期的登记（当班进程做）：租约过期超过十倍租约时长的。它们认领的行早已被接手。""" from app.db.models_cluster import ClusterWorker - floor = (now or utcnow()) - timedelta(seconds=10 * lease_s()) async with SessionLocal() as session: + floor = (now if now is not None else clock(session)) - timedelta(seconds=10 * lease_s()) done = await session.execute(delete(ClusterWorker).where(ClusterWorker.lease_until < floor)) await session.commit() return int(done.rowcount or 0) @@ -150,8 +174,8 @@ async def listing() -> list[dict[str, Any]]: """现在登记着的进程（系统信息页、验收脚本看）。""" from app.db.models_cluster import ClusterWorker - now = utcnow() async with SessionLocal() as session: + now = await db_now(session) rows = (await session.execute(select(ClusterWorker).order_by(ClusterWorker.started_at, ClusterWorker.id))).scalars().all() return [{"id": r.id, "host": r.host, "pid": r.pid, "alive": r.lease_until > now, "me": r.id == identity.owner(), "code_version": r.code_version, "schema_version": r.schema_version, @@ -159,5 +183,5 @@ async def listing() -> list[dict[str, Any]]: "heartbeat_at": r.heartbeat_at.isoformat() if r.heartbeat_at else None} for r in rows] -__all__ = ["Liveness", "alive", "heartbeat_s", "lease_s", "listing", "liveness", "prune", "register", "renew", - "unregister"] +__all__ = ["Liveness", "alive", "clock", "db_now", "heartbeat_s", "lease_s", "listing", "liveness", "prune", "register", + "renew", "unregister"] diff --git a/backend/tests/test_cluster.py b/backend/tests/test_cluster.py index 89c708cd..b92a68fd 100644 --- a/backend/tests/test_cluster.py +++ b/backend/tests/test_cluster.py @@ -355,6 +355,70 @@ async def test_leadership_is_handed_over_when_the_lease_runs_out(): assert (await leader_mod.current("test-lead"))["owner"] is None +def test_the_fence_counts_what_the_monotonic_clock_missed(monkeypatch): + """复核 R-b7：单调钟不走（休眠不计、虚拟机挂起）而墙钟过了两个租约——照样算续晚了；两只钟差出一个租约算跳了。""" + from app.cluster import fence + + clocks = {"mono": 1000.0, "wall": 5000.0} + monkeypatch.setattr(fence, "_clock", lambda: clocks["mono"]) + monkeypatch.setattr(fence, "_wall", lambda: clocks["wall"]) + fence.activate() + try: + assert not fence.stale() and not fence.jumped() + clocks["wall"] += 2 * workers.lease_s() # 合盖两个租约：单调钟没计，墙钟走了 + assert fence.stale() and fence.jumped() + fence.renewed() + clocks["mono"] += 2 * workers.lease_s() # 反过来：墙钟被往回拨了，单调钟照走 + assert fence.stale() and fence.jumped() + fence.renewed() + clocks["mono"] += workers.lease_s() / 2 + clocks["wall"] += workers.lease_s() / 2 + assert not fence.stale() and not fence.jumped() + finally: + fence.deactivate() + + +@pytest.mark.skipif(app_settings.db_dialect != "postgresql", reason="SQLite 只有一个进程，租约用本机的钟") +async def test_leases_use_the_database_clock_not_the_local_wall_clock(monkeypatch): + """复核 R-b7：本机墙钟快了一个小时——别的进程照样算活着、它拿着的当班租约照样抢不走；按库的时钟自己的租约已经过期时，续租 + 不再续这个身份，而是隔离。""" + from app.cluster import fence + from app.cluster.service import service + from app.db.base import utcnow as real_utcnow + import app.db.base as db_base + + await _peer(PEER) + lead = leader_mod.Leadership("test-skew") + async with SessionLocal() as session: + session.add(ClusterLease(name="test-skew", owner=PEER, lease_until=real_utcnow() + timedelta(seconds=60), epoch=1)) + await session.commit() + skewed = lambda: real_utcnow() + timedelta(hours=1) # noqa: E731 + monkeypatch.setattr(workers, "utcnow", skewed) + monkeypatch.setattr(db_base, "utcnow", skewed) + assert PEER in (await workers.liveness()).others, "按库的时钟它的租约没过期" + assert not await lead.claim(), "当班租约按库的时钟判：抢不走" + async with SessionLocal() as session: + await session.execute(delete(ClusterWorker).where(ClusterWorker.id == PEER)) # 它没记版本：登记时会被版本守卫拦下 + await session.execute(delete(ClusterLease).where(ClusterLease.name == "test-skew")) + await session.commit() + + await workers.register() + fence.activate() + before = fence.generation() + try: + assert await service.renew(), "按库的时钟没过期：续上" + async with SessionLocal() as session: + row = await session.get(ClusterWorker, identity.owner()) + row.lease_until = real_utcnow() - timedelta(seconds=1) # 库的时钟已经过了它的租约，本机的两只钟都没觉得 + await session.commit() + assert not await service.renew() and fence.generation() == before + 1, "过期了不续这个身份：隔离" + finally: + fence.deactivate() + if service._lease_conn is not None: + await service._lease_conn.close() + service._lease_conn = None + + async def test_the_daily_cap_is_not_overrun_by_concurrent_charges(): from app.engine.spend import Ledger diff --git a/backend/tests/test_cluster_fence.py b/backend/tests/test_cluster_fence.py index f963e9c6..b48aaba6 100644 --- a/backend/tests/test_cluster_fence.py +++ b/backend/tests/test_cluster_fence.py @@ -99,7 +99,7 @@ async def _stall_and_take_over(slow) -> str: await asyncio.wait_for(slow["entered"].wait(), timeout=15) await _checkpointed_before_slow(run.id) before = fence.generation() - fence.renewed(time.monotonic() - lease_s() - 1) # 卡了一个多租约时长 + fence.renewed(ago=lease_s() + 1) # 卡了一个多租约时长 await _taken_over(run.id) slow["gate"].set() # 缓过来，手上的执行接着往下走 for _ in range(200):