mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 09:31:59 +00:00
复核 R-b7(休眠、时钟偏差下隔离失灵):租约改用库的时钟——续租、登记写 lease_until = 库的 clock_timestamp() + 租约,判活、清理过期登记、版本守卫看谁活着都比库的时钟,调度当班的租约同样按库的时钟接班、续租,本地判还当不当班改看计入休眠的单调钟,机器之间墙钟差多少都不起作用;隔离改用计入休眠的钟(Linux CLOCK_BOOTTIME,macOS clock_gettime(CLOCK_MONOTONIC),取不到退回 monotonic),并且一律同时比墙钟,两只钟里走得多的那只说过了一个租约就算续晚了;续租时两只钟差出一个租约(墙钟跳了、休眠没计)也隔离;续租那一句带上「库里自己的租约还没过期」的条件,过期了或者登记不见了就不再续这个身份、隔离换身份(以前登记不见了是用旧身份重新登记)。补测试:单调钟不走而墙钟过了两个租约、墙钟往回拨,都算续晚了;本机墙钟快一小时,别的进程照样活着、当班租约抢不走;库的时钟过了自己的租约时续租不续、隔离一次
This commit is contained in:
@@ -1188,7 +1188,7 @@ asyncpg 方言一律这样做,同一个事务里、在同一条服务端连接
|
||||
SQL 形状。各进程各自的并发闸门(口径比对、自动对比、起草的模型调用、上传解析)按进程算,进程多了总量跟着多。
|
||||
|
||||
**多台机器。** 几台机器要用同一份数据目录:放在共享存储(NFS 之类)上,每台挂在**同一个绝对路径**(库里记着上传表格各版本库文件的
|
||||
绝对路径),各台机器的时钟要对时(NTP;租约按墙钟算)。文件都是先写临时文件再改名,读的一方看不到写了一半的;写上传版本库、记录表
|
||||
绝对路径),各台机器的时钟仍要对时(NTP):租约按库的时钟算,机器之间差多少都不影响谁算活着,但「停了两个租约时长」这类判断看的是各机器按本机时钟写下的时间。文件都是先写临时文件再改名,读的一方看不到写了一半的;写上传版本库、记录表
|
||||
在库里的锁下一个接一个。但要说实话:数据目录里还有 SQLite 文件——上传表格的各版本库(发布后只读)、填报的记录表(`records/*.db`,
|
||||
原地追加写)、沙箱里代码自己建的库——SQLite 文件放在网络文件系统上,读写一致性靠文件锁,而 NFS 上的文件锁不可靠:一台机器在追加时
|
||||
另一台读,可能读到不一致的页。所以多台机器时,记录表写得多的部署建议用一台机器多进程,或者把数据目录放在支持可靠文件锁的存储上。
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)]
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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):
|
||||
|
||||
Reference in New Issue
Block a user