From eb9af51abb63b242c5bb5d53cced9d400c4d57c5 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 15:07:17 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=8D=E6=A0=B8=20R-b14=EF=BC=88PostgreSQL?= =?UTF-8?q?=20=E5=8D=95=E8=BF=9B=E7=A8=8B=E5=A2=99=E9=92=9F=E4=B8=80?= =?UTF-8?q?=E8=B7=B3=E5=B0=B1=E9=9A=94=E7=A6=BB=E8=87=AA=E5=B7=B1=EF=BC=89?= =?UTF-8?q?=EF=BC=9A=E5=8F=96=E5=BE=97=E5=88=B0=E8=AE=A1=E5=85=A5=E4=BC=91?= =?UTF-8?q?=E7=9C=A0=E7=9A=84=E9=92=9F=EF=BC=88SLEEP=5FAWARE=EF=BC=89?= =?UTF-8?q?=E6=97=B6=20stale()=20=E5=8F=AA=E7=9C=8B=E5=AE=83=EF=BC=8C?= =?UTF-8?q?=E5=A2=99=E9=92=9F=E5=8F=AA=E5=9C=A8=E5=8F=96=E4=B8=8D=E5=88=B0?= =?UTF-8?q?=E6=97=B6=E5=BD=93=E5=90=8E=E5=A4=87=EF=BC=88=E4=B8=A4=E5=8F=AA?= =?UTF-8?q?=E9=87=8C=E8=B5=B0=E5=BE=97=E5=A4=9A=E7=9A=84=E9=82=A3=E5=8F=AA?= =?UTF-8?q?=EF=BC=89=EF=BC=9B=E4=B8=A4=E5=8F=AA=E9=92=9F=E5=AF=B9=E4=B8=8D?= =?UTF-8?q?=E4=B8=8A=EF=BC=88jumped=EF=BC=89=E4=B8=8D=E5=86=8D=E7=9B=B4?= =?UTF-8?q?=E6=8E=A5=E9=9A=94=E7=A6=BB=EF=BC=8C=E5=85=88=E7=BB=AD=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=E7=A7=9F=E2=80=94=E2=80=94=E7=BB=AD=E7=A7=9F=E9=82=A3?= =?UTF-8?q?=E4=B8=80=E5=8F=A5=E6=8C=89=E5=BA=93=E7=9A=84=E6=97=B6=E9=92=9F?= =?UTF-8?q?=E6=A0=B8=E5=AF=B9=E3=80=8C=E8=87=AA=E5=B7=B1=E7=9A=84=E7=A7=9F?= =?UTF-8?q?=E7=BA=A6=E8=BF=98=E6=B2=A1=E8=BF=87=E6=9C=9F=E3=80=8D=EF=BC=8C?= =?UTF-8?q?=E4=BB=A5=E5=AE=83=E4=B8=BA=E5=87=86=EF=BC=9A=E7=BB=AD=E4=B8=8A?= =?UTF-8?q?=E4=BA=86=E5=8F=AA=E8=AE=B0=E4=B8=80=E6=9D=A1=E6=97=A5=E5=BF=97?= =?UTF-8?q?=E7=85=A7=E5=B8=B8=EF=BC=8C=E7=BB=AD=E4=B8=8D=E4=B8=8A=E6=89=8D?= =?UTF-8?q?=E9=9A=94=E7=A6=BB=E3=80=82README=E3=80=8C=E5=A4=9A=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E3=80=8D=E8=A1=A5=E4=B8=80=E5=8F=A5=EF=BC=9A=E5=BA=93?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E5=99=A8=E8=87=AA=E5=B7=B1=E7=9A=84=E6=97=B6?= =?UTF-8?q?=E9=92=9F=E5=BE=80=E5=89=8D=E8=B7=B3=E8=B6=85=E8=BF=87=E4=B8=80?= =?UTF-8?q?=E4=B8=AA=E7=A7=9F=E7=BA=A6=E6=97=B6=E6=89=80=E6=9C=89=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E4=B8=80=E8=B5=B7=E9=9A=94=E7=A6=BB=E3=80=81=E6=80=8E?= =?UTF-8?q?=E4=B9=88=E6=81=A2=E5=A4=8D=EF=BC=8C=E5=BE=80=E5=9B=9E=E8=B7=B3?= =?UTF-8?q?=E6=97=B6=E7=9C=9F=E6=B2=A1=E4=BA=86=E7=9A=84=E8=BF=9B=E7=A8=8B?= =?UTF-8?q?=E8=A6=81=E5=A4=9A=E7=AD=89=EF=BC=8C=E6=9C=AC=E6=9C=BA=E5=A2=99?= =?UTF-8?q?=E9=92=9F=E8=B7=B3=E4=BA=86=E4=B8=8D=E9=9A=94=E7=A6=BB=E3=80=82?= =?UTF-8?q?=E8=A1=A5=E6=B5=8B=E8=AF=95=EF=BC=9A=E5=A2=99=E9=92=9F=E5=BE=80?= =?UTF-8?q?=E5=89=8D=E5=BE=80=E5=90=8E=E8=B7=B3=E4=B8=80=E5=B0=8F=E6=97=B6?= =?UTF-8?q?=E4=B8=8D=E7=AE=97=E7=BB=AD=E6=99=9A=E4=BA=86=E3=80=81=E5=8F=AA?= =?UTF-8?q?=E7=AE=97=E5=AF=B9=E4=B8=8D=E4=B8=8A=EF=BC=8C=E8=AE=A1=E5=85=A5?= =?UTF-8?q?=E4=BC=91=E7=9C=A0=E7=9A=84=E9=92=9F=E8=B5=B0=E8=BF=87=E4=B8=A4?= =?UTF-8?q?=E4=B8=AA=E7=A7=9F=E7=BA=A6=E7=85=A7=E6=A0=B7=E7=AE=97=EF=BC=8C?= =?UTF-8?q?=E5=8F=96=E4=B8=8D=E5=88=B0=E6=97=B6=E5=A2=99=E9=92=9F=E5=BD=93?= =?UTF-8?q?=E5=90=8E=E5=A4=87=EF=BC=9BPostgreSQL=20=E5=8D=95=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E5=A2=99=E9=92=9F=20=C2=B11=20=E5=B0=8F=E6=97=B6?= =?UTF-8?q?=E7=BB=AD=E7=A7=9F=E7=85=A7=E5=B8=B8=E7=BB=AD=E4=B8=8A=E3=80=81?= =?UTF-8?q?=E4=B8=8D=E9=9A=94=E7=A6=BB=EF=BC=8C=E7=9C=9F=E8=BF=87=E6=9C=9F?= =?UTF-8?q?=E4=BB=8D=E9=9A=94=E7=A6=BB=EF=BC=9B=E6=89=A7=E8=A1=8C=E5=88=B0?= =?UTF-8?q?=E4=B8=80=E5=8D=8A=E5=A2=99=E9=92=9F=20=C2=B11=20=E5=B0=8F?= =?UTF-8?q?=E6=97=B6=EF=BC=8C=E8=BF=90=E8=A1=8C=E7=85=A7=E5=B8=B8=E8=B7=91?= =?UTF-8?q?=E5=AE=8C=E3=80=81=E4=B8=96=E4=BB=A3=E4=B8=8D=E5=8F=98=EF=BC=88?= =?UTF-8?q?=E6=94=B9=E5=89=8D=E7=9A=84=E5=86=99=E6=B3=95=E5=BE=80=E5=89=8D?= =?UTF-8?q?=E8=B7=B3=E9=82=A3=E6=9D=A1=E4=B8=8D=E8=BF=87=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 +- backend/app/cluster/fence.py | 21 +++++++++++--------- backend/app/cluster/service.py | 14 +++++++++----- backend/tests/test_cluster.py | 30 +++++++++++++++++++++++------ backend/tests/test_cluster_fence.py | 21 ++++++++++++++++++++ 5 files changed, 67 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index 35e4e194..1869eda3 100644 --- a/README.md +++ b/README.md @@ -1188,7 +1188,7 @@ asyncpg 方言一律这样做，同一个事务里、在同一条服务端连接 SQL 形状。各进程各自的并发闸门（口径比对、自动对比、起草的模型调用、上传解析）按进程算，进程多了总量跟着多。 **多台机器。** 几台机器要用同一份数据目录：放在共享存储（NFS 之类）上，每台挂在**同一个绝对路径**（库里记着上传表格各版本库文件的 -绝对路径），各台机器的时钟仍要对时（NTP）：租约按库的时钟算，机器之间差多少都不影响谁算活着，但「停了两个租约时长」这类判断看的是各机器按本机时钟写下的时间。文件都是先写临时文件再改名，读的一方看不到写了一半的；写上传版本库、记录表 +绝对路径），各台机器的时钟仍要对时（NTP）：租约按库的时钟算，机器之间差多少都不影响谁算活着，但「停了两个租约时长」这类判断看的是各机器按本机时钟写下的时间。库服务器自己的时钟往前跳超过一个租约时（少见），所有进程续租时都会发现自己的租约已经过期、一起隔离：执行到一半的运行标成中断，在运行详情里点「继续运行」从断点接着执行；后台任务按「属主不在了」的规矩处理；各进程自动换身份重新登记，之后发起的照常。往回跳时不隔离，只是真没了的进程要多等跳回去的那么久才被接手。本机墙钟跳了（NTP 一步校时、手动改时间）不会隔离：判续没续晚看的是计入休眠的单调钟，对不上时以库里的租约为准。文件都是先写临时文件再改名，读的一方看不到写了一半的；写上传版本库、记录表 在库里的锁下一个接一个。但要说实话：数据目录里还有 SQLite 文件——上传表格的各版本库（发布后只读）、填报的记录表（`records/*.db`， 原地追加写）、沙箱里代码自己建的库——SQLite 文件放在网络文件系统上，读写一致性靠文件锁，而 NFS 上的文件锁不可靠：一台机器在追加时 另一台读，可能读到不一致的页。所以多台机器时，记录表写得多的部署建议用一台机器多进程，或者把数据目录放在支持可靠文件锁的存储上。 diff --git a/backend/app/cluster/fence.py b/backend/app/cluster/fence.py index 1d94b3d6..ec1eecc7 100644 --- a/backend/app/cluster/fence.py +++ b/backend/app/cluster/fence.py @@ -11,12 +11,14 @@ - 一旦发现续晚了（trip）：世代加一，手上的执行全部停掉（停的时候不写终态，那些行交给接手的进程按「属主不在了」处理），换一个新的 身份重新登记；新身份登记上之前新开始的执行先等着（ready），登记上之后照常。 -**用哪只钟（复核 R-b7）。** 别的进程判「租约过期」用的是库的时钟（cluster/workers：续租写 lease_until = 库的 now() + 租约）， -跟各台机器的墙钟差多少无关。这边判「续晚了」用计入休眠的钟：time.monotonic() 在 macOS（mach_absolute_time）、Linux +**用哪只钟（复核 R-b7、R-b14）。** 别的进程判「租约过期」用的是库的时钟（cluster/workers：续租写 lease_until = 库的 now() + +租约），跟各台机器的墙钟差多少无关。这边判「续晚了」用计入休眠的钟：time.monotonic() 在 macOS（mach_absolute_time）、Linux （CLOCK_MONOTONIC）上都不计休眠——笔记本合盖、虚拟机挂起超过两个租约，对面早已接手，这边还以为刚续过。所以 Linux 用 -CLOCK_BOOTTIME，macOS 用 clock_gettime(CLOCK_MONOTONIC)（这个计入休眠），都取不到时退回 monotonic；不论哪种，另外同时比墙钟， -两只钟里走得多的那只说过了一个租约就算续晚了。续租时再比一下两只钟（jumped）：差出一个租约以上（墙钟跳了、或者计入休眠的钟 -其实没计）就隔离。续租那一句本身也按库的时钟核对：库里自己的租约已经过期（或者登记不见了），就不再续这个身份。 +CLOCK_BOOTTIME，macOS 用 clock_gettime(CLOCK_MONOTONIC)（这个计入休眠），只看它；墙钟会被 NTP 一步校时、手动改时间、虚拟机 +对时拨来拨去，取得到计入休眠的钟时不拿它判（否则单进程墙钟一跳就把自己隔离了）。两只都取不到时退回 monotonic，墙钟当后备： +两只里走得多的那只说过了一个租约就算续晚了。两只钟对不上（jumped，差出一个租约以上）时不直接隔离：先续一次租，续租那一句按库的 +时钟核对「自己的租约还没过期」，以它为准——续上了记一条日志照常，续不上（库里自己的租约已经过期、或者登记不见了）才隔离。 +虚拟机被宿主挂起时客户机的 CLOCK_BOOTTIME 也可能停着：那时靠续租那一句（最迟一拍心跳之后）发现过期。 Fenced 是 BaseException：节点、工具里常见的 except Exception 拦不住它，一路退到执行的最外层。没起协调（测试、脚本）时什么都不查。 """ @@ -98,8 +100,8 @@ def bind() -> None: def stale() -> bool: - """离上次续上租约是不是已经超过一个租约时长：两只钟（计入休眠的单调钟、墙钟）里走得多的那只说了算。隔离之后、新身份还没 - 登记上的那一段也算（pending）。""" + """离上次续上租约是不是已经超过一个租约时长：取得到计入休眠的钟时只看它；取不到时它（monotonic）和墙钟里走得多的那只说了算。 + 隔离之后、新身份还没登记上的那一段也算（pending）。""" if not _active: return False since = _since() @@ -107,7 +109,7 @@ def stale() -> bool: return True from app.cluster.workers import lease_s - return max(since) > lease_s() + return (since[0] if SLEEP_AWARE else max(since)) > lease_s() def pending() -> bool: @@ -131,7 +133,8 @@ async def ready(timeout: float | None = None) -> None: def jumped() -> bool: - """两只钟差出一个租约以上（墙钟往前或往后跳了，或者单调钟没计入休眠）：续租时问，是就隔离。""" + """两只钟差出一个租约以上（墙钟往前或往后跳了，或者单调钟没计入休眠）。续租时问：是的话不直接隔离，以续租那一句按库的时钟 + 核对的结果为准（cluster/service.renew）。""" if not _active: return False since = _since() diff --git a/backend/app/cluster/service.py b/backend/app/cluster/service.py index 4bb6cf29..2d40ed08 100644 --- a/backend/app/cluster/service.py +++ b/backend/app/cluster/service.py @@ -131,15 +131,15 @@ class _Service: async def renew(self) -> bool: """续一拍租约（在专用连接上，不走连接池：池子被占满时取连接要等，等得比租约还长）。按库的时钟续、按库的时钟核对（复核 R-b7）：库里自己的租约已经过期（或者登记不见了），别的进程可能已经接手，不再续这个身份——隔离（cluster/fence.py），换身份 - 重新登记。本地的两只钟说续晚了、或者两只钟差出一个租约（墙钟跳了），同样隔离。交回续没续上。""" + 重新登记。计入休眠的钟说续晚了，同样隔离；两只钟差出一个租约（墙钟跳了）只记一条日志，以库的核对为准（复核 R-b14）。 + 交回续没续上。""" from app.cluster import fence if fence.stale(): fence.trip("续租约的时候发现已经续晚了") return False - if fence.jumped(): - fence.trip("续租约的时候发现本机的墙钟和单调钟差出一个租约以上（墙钟跳了，或者休眠没计进单调钟）") - return False + # 两只钟对不上（墙钟跳了，或者单调钟没计入休眠）：不直接隔离，以下面这一句按库的时钟核对的结果为准（复核 R-b14） + jumped = fence.jumped() try: conn = await self._lease_conn_open() renewed = await asyncio.wait_for(conn.fetchval( @@ -159,8 +159,12 @@ class _Service: self._lease_conn = None return False if renewed is None: - fence.trip("按库的时钟本进程的租约已经过期" if expired else "本进程的登记不见了（被当成不在了）") + why = "按库的时钟本进程的租约已经过期" if expired else "本进程的登记不见了（被当成不在了）" + fence.trip(why + ("；本机的墙钟和单调钟也差出了一个租约以上" if jumped else "")) return False + if jumped: + logger.warning("本机的墙钟和单调钟差出一个租约以上（墙钟跳了：NTP 校时、手动改时间、虚拟机对时），按库的时钟本进程的租约" + "还没过期：照常续上，不隔离") fence.renewed() return True diff --git a/backend/tests/test_cluster.py b/backend/tests/test_cluster.py index ece95be2..8b7aea5d 100644 --- a/backend/tests/test_cluster.py +++ b/backend/tests/test_cluster.py @@ -7,6 +7,7 @@ import asyncio import os import subprocess import sys +import time from datetime import timedelta from pathlib import Path @@ -412,25 +413,34 @@ 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：单调钟不走（休眠不计、虚拟机挂起）而墙钟过了两个租约——照样算续晚了；两只钟差出一个租约算跳了。""" +def test_the_fence_judges_by_the_sleep_aware_clock_and_falls_back_to_the_wall(monkeypatch): + """复核 R-b7、R-b14：取得到计入休眠的钟时只看它——它走过两个租约（合盖、挂起）算续晚了，墙钟往前、往后跳一小时不算（只算 + 两只钟对不上，jumped）；取不到计入休眠的钟时，monotonic 和墙钟里走得多的那只说了算（墙钟当后备）。""" 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"]) + monkeypatch.setattr(fence, "SLEEP_AWARE", True) fence.activate() try: assert not fence.stale() and not fence.jumped() - clocks["wall"] += 2 * workers.lease_s() # 合盖两个租约：单调钟没计，墙钟走了 - assert fence.stale() and fence.jumped() + for jump in (3600.0, -3600.0): # NTP 一步校时、手动改时间 + fence.renewed() + clocks["wall"] += jump + assert not fence.stale() and fence.jumped(), "只是墙钟跳了：不算续晚了" fence.renewed() - clocks["mono"] += 2 * workers.lease_s() # 反过来：墙钟被往回拨了，单调钟照走 - assert fence.stale() and fence.jumped() + clocks["mono"] += 2 * workers.lease_s() # 计入休眠的钟走过两个租约（合盖） + assert fence.stale() fence.renewed() clocks["mono"] += workers.lease_s() / 2 clocks["wall"] += workers.lease_s() / 2 assert not fence.stale() and not fence.jumped() + + monkeypatch.setattr(fence, "SLEEP_AWARE", False) # 取不到计入休眠的钟：monotonic 不计休眠，墙钟当后备 + fence.renewed() + clocks["wall"] += 2 * workers.lease_s() + assert fence.stale() finally: fence.deactivate() @@ -464,6 +474,14 @@ async def test_leases_use_the_database_clock_not_the_local_wall_clock(monkeypatc before = fence.generation() try: assert await service.renew(), "按库的时钟没过期：续上" + monkeypatch.setattr(fence, "SLEEP_AWARE", True) + for jump in (3600.0, -3600.0): # 单进程、墙钟往前往后跳一小时（复核 R-b14） + monkeypatch.setattr(fence, "_wall", lambda jump=jump: time.time() + jump) + assert not fence.stale() and fence.jumped() + fence.check() + assert await service.renew() and fence.generation() == before, "库的时钟说租约还在：续上、不隔离" + assert not fence.jumped(), "续上之后两只钟重新对齐" + monkeypatch.setattr(fence, "_wall", time.time) async with SessionLocal() as session: row = await session.get(ClusterWorker, identity.owner()) row.lease_until = real_utcnow() - timedelta(seconds=1) # 库的时钟已经过了它的租约，本机的两只钟都没觉得 diff --git a/backend/tests/test_cluster_fence.py b/backend/tests/test_cluster_fence.py index 906db81e..b40ebd64 100644 --- a/backend/tests/test_cluster_fence.py +++ b/backend/tests/test_cluster_fence.py @@ -149,6 +149,27 @@ async def test_a_run_continued_elsewhere_finishes_once_and_the_stalled_side_stay await _no_repeated_seq(run_id) +@pytest.mark.parametrize("jump", [3600.0, -3600.0]) +async def test_a_wall_clock_jump_does_not_fence_a_running_run(slow, monkeypatch, jump): + """复核 R-b14：执行到一半本机墙钟往前 / 往后跳一小时（NTP 一步校时、手动改时间）——计入休眠的钟没问题，不隔离，运行照常跑完。""" + import time as _time + + monkeypatch.setattr(fence, "SLEEP_AWARE", True) + run = await run_manager.start(graph=GRAPH, input_payload={"question": "q"}) + await asyncio.wait_for(slow["entered"].wait(), timeout=15) + before = fence.generation() + monkeypatch.setattr(fence, "_wall", lambda: _time.time() + jump) + slow["gate"].set() + for _ in range(300): + async with SessionLocal() as session: + row = await session.get(Run, run.id) + if row.status in ("succeeded", "failed", "interrupted") and not run_manager.is_active(run.id): + break + await asyncio.sleep(0.05) + assert row.status == "succeeded", row.error + assert fence.generation() == before, "没有隔离" + + 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={})