From 68c1f4143ec6d7822abf2844ec5a99288ad85cba Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 14:45:54 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=8D=E6=A0=B8=20R-b13=EF=BC=88=E7=89=88?= =?UTF-8?q?=E6=9C=AC=E6=8B=A6=E6=88=AA=E4=B8=8E=E9=9B=B6=E7=A2=8E=EF=BC=89?= =?UTF-8?q?=EF=BC=9Apreflight=20=E6=8A=A5=E9=94=99=E5=88=97=E5=87=BA?= =?UTF-8?q?=E5=85=A8=E9=83=A8=E6=B4=BB=E7=9D=80=E7=9A=84=E3=80=81=E7=89=88?= =?UTF-8?q?=E6=9C=AC=E4=B8=8D=E5=90=8C=E7=9A=84=E8=BF=9B=E7=A8=8B=EF=BC=8C?= =?UTF-8?q?=E5=90=84=E6=A0=87=E6=98=8E=E5=BF=83=E8=B7=B3=E5=9C=A8=E8=B5=B0?= =?UTF-8?q?=E8=BF=98=E6=98=AF=E5=81=9C=E7=9D=80=E3=80=81=E7=AD=89=E7=A7=9F?= =?UTF-8?q?=E7=BA=A6=E8=BF=87=E6=9C=9F=EF=BC=88=E4=BB=A5=E5=89=8D=E5=8F=AA?= =?UTF-8?q?=E7=82=B9=E5=90=8D=E5=85=88=E7=9C=8B=E5=88=B0=E5=BF=83=E8=B7=B3?= =?UTF-8?q?=E7=9A=84=E9=82=A3=E4=B8=AA=EF=BC=8C=E4=BA=BA=E5=81=9C=E6=8E=89?= =?UTF-8?q?=E5=AE=83=E5=86=8D=E8=B5=B7=E8=BF=98=E4=BC=9A=E8=A2=AB=E5=8F=A6?= =?UTF-8?q?=E4=B8=80=E4=B8=AA=E6=8B=A6=E4=B8=8B=EF=BC=89=EF=BC=9B=E6=BA=90?= =?UTF-8?q?=E7=A0=81=E6=8C=87=E7=BA=B9=E6=8A=8A=E4=BE=9D=E8=B5=96=E9=94=81?= =?UTF-8?q?=E6=96=87=E4=BB=B6=20requirements.lock=20=E7=9A=84=E5=86=85?= =?UTF-8?q?=E5=AE=B9=E4=B9=9F=E7=AE=97=E8=BF=9B=E5=8E=BB=EF=BC=8C=E5=8F=AA?= =?UTF-8?q?=E5=8D=87=E4=BA=86=E4=BE=9D=E8=B5=96=E4=B9=9F=E7=AE=97=E7=89=88?= =?UTF-8?q?=E6=9C=AC=E4=B8=8D=E5=90=8C=EF=BC=8C=E9=94=81=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E5=8F=96=E4=B8=8D=E5=88=B0=E6=97=B6=E6=8D=A2=E6=88=90=E5=87=A0?= =?UTF-8?q?=E4=B8=AA=E5=85=B3=E9=94=AE=E5=8C=85=E7=9A=84=E7=89=88=E6=9C=AC?= =?UTF-8?q?=EF=BC=8CDocker=20=E9=95=9C=E5=83=8F=E9=87=8C=E7=95=99=E4=B8=80?= =?UTF-8?q?=E4=BB=BD=E9=94=81=E6=96=87=E4=BB=B6=EF=BC=9B=E5=AD=98=E9=87=8F?= =?UTF-8?q?=E5=BA=93=20run=5Fevents=20=E6=92=9E=E4=BA=86=E5=8F=B7=E6=97=B6?= =?UTF-8?q?=E6=8A=8A=E7=BB=93=E6=9E=9C=E8=AE=B0=E4=B8=80=E6=AC=A1=EF=BC=88?= =?UTF-8?q?settings=20=E8=A1=A8=20cluster:run=5Fevents=5Fseq=5Fduplicates?= =?UTF-8?q?=EF=BC=8C=E8=AE=B0=E4=B8=8B=E6=92=9E=E5=8F=B7=E7=9A=84=E9=82=A3?= =?UTF-8?q?=E5=87=A0=E5=A4=84=EF=BC=8C=E6=9C=80=E5=A4=9A=20200=20=E5=A4=84?= =?UTF-8?q?=EF=BC=89=EF=BC=8C=E4=B9=8B=E5=90=8E=E5=90=AF=E5=8A=A8=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E6=95=B4=E8=A1=A8=20GROUP=20BY=E3=80=81=E5=8F=AA?= =?UTF-8?q?=E6=8C=89=E7=B4=A2=E5=BC=95=E6=A0=B8=E5=AF=B9=E9=82=A3=E5=87=A0?= =?UTF-8?q?=E5=A4=84=E8=BF=98=E6=92=9E=E4=B8=8D=E6=92=9E=EF=BC=8C=E9=83=BD?= =?UTF-8?q?=E5=A4=84=E7=90=86=E6=8E=89=E4=BA=86=E6=89=8D=E5=86=8D=E6=95=B0?= =?UTF-8?q?=E4=B8=80=E9=81=8D=E3=80=81=E5=BB=BA=E5=94=AF=E4=B8=80=E7=B4=A2?= =?UTF-8?q?=E5=BC=95=E3=80=81=E5=88=A0=E6=8E=89=E8=AE=B0=E5=BD=95=E3=80=82?= =?UTF-8?q?=E9=9A=94=E7=A6=BB=E5=90=8E=E6=8D=A2=E8=BA=AB=E4=BB=BD=E9=82=A3?= =?UTF-8?q?=E5=87=A0=E6=9D=A1=E6=97=A5=E5=BF=97=E5=86=99=E5=87=BA=E6=96=B0?= =?UTF-8?q?=E6=97=A7=E5=AE=8C=E6=95=B4=E8=BA=AB=E4=BB=BD=E5=B7=B2=E5=9C=A8?= =?UTF-8?q?=20R-b9=20=E9=82=A3=E4=B8=80=E7=AC=94=E9=87=8C=E6=94=B9?= =?UTF-8?q?=E4=BA=86=E3=80=82README=E3=80=8C=E5=8D=87=E7=BA=A7=E3=80=8D?= =?UTF-8?q?=E4=B8=80=E6=AE=B5=E8=B7=9F=E7=9D=80=E8=A1=A5=E4=B8=80=E5=8F=A5?= =?UTF-8?q?=E3=80=82=E8=A1=A5=E6=B5=8B=E8=AF=95=EF=BC=9A=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=E5=BF=83=E8=B7=B3=E5=9C=A8=E8=B5=B0=E3=80=81=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=E5=BF=83=E8=B7=B3=E5=81=9C=E7=9D=80=EF=BC=8C=E6=8A=A5=E9=94=99?= =?UTF-8?q?=E4=B8=A4=E4=B8=AA=E9=83=BD=E5=88=97=E5=87=BA=E3=80=81=E5=90=84?= =?UTF-8?q?=E6=A0=87=E7=8A=B6=E6=80=81=EF=BC=9B=E9=94=81=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E5=8F=98=E4=BA=86=E6=8C=87=E7=BA=B9=E8=B7=9F=E7=9D=80=E5=8F=98?= =?UTF-8?q?=E3=80=81=E5=8F=96=E4=B8=8D=E5=88=B0=E6=97=B6=E7=94=A8=E5=8C=85?= =?UTF-8?q?=E7=89=88=E6=9C=AC=EF=BC=9B=E6=92=9E=E5=8F=B7=E7=AC=AC=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=E6=95=B4=E8=A1=A8=E6=95=B0=E3=80=81=E7=AC=AC=E4=BA=8C?= =?UTF-8?q?=E6=AC=A1=E5=8F=AA=E6=A0=B8=E5=AF=B9=E4=B8=8D=E6=89=AB=E3=80=81?= =?UTF-8?q?=E5=A4=84=E7=90=86=E6=8E=89=E4=B9=8B=E5=90=8E=E5=BB=BA=E4=B8=8A?= =?UTF-8?q?=E7=B4=A2=E5=BC=95=E5=B9=B6=E5=88=A0=E6=8E=89=E8=AE=B0=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Dockerfile | 2 + README.md | 4 +- backend/app/cluster/version.py | 47 ++++++++++++--- backend/app/db/base.py | 79 +++++++++++++++++++++++-- backend/tests/test_cluster.py | 103 +++++++++++++++++++++++++++++++++ 5 files changed, 220 insertions(+), 15 deletions(-) diff --git a/Dockerfile b/Dockerfile index c3bd9ba7..54359a04 100644 --- a/Dockerfile +++ b/Dockerfile @@ -76,6 +76,8 @@ SH WORKDIR /app COPY LICENSE COMMERCIAL-LICENSE.md /app/ COPY backend/app /app/backend/app +# 锁文件留一份：进程登记的代码版本把它算进源码指纹（app/cluster/version.py：升了依赖的进程也算版本不同） +COPY backend/requirements.lock /app/backend/requirements.lock COPY --from=web /src/frontend/dist /app/web ENV AGENTLAB_WEB_DIST=/app/web \ diff --git a/README.md b/README.md index e183bce6..35e4e194 100644 --- a/README.md +++ b/README.md @@ -1194,8 +1194,8 @@ SQL 形状。各进程各自的并发闸门（口径比对、自动对比、起 另一台读，可能读到不一致的页。所以多台机器时，记录表写得多的部署建议用一台机器多进程，或者把数据目录放在支持可靠文件锁的存储上。 沙箱会话（microVM）在各进程自己手上，不跨进程共享。 -**升级。** 不支持新旧版本的进程同时连着一个库，代码里拦着：每个进程登记时带上代码版本（包版本、构建号或 git 提交、源码指纹）和 -库结构版本，库里还有活着的、版本不同的进程时，新版本的进程在补列之前就拒绝启动，报错里写明是哪几个进程、什么版本；同版本的照常 +**升级。** 不支持新旧版本的进程同时连着一个库，代码里拦着：每个进程登记时带上代码版本（包版本、构建号或 git 提交、源码指纹——源码加上依赖锁文件，只升了依赖也算版本不同）和 +库结构版本，库里还有活着的、版本不同的进程时，新版本的进程在补列之前就拒绝启动，报错里列出全部这样的进程、各是什么版本、心跳还在不在走；同版本的照常 加入。升级步骤： 1. 停掉全部旧版本的进程（平稳关停的马上删掉自己的登记；强杀的等它的租约过期，新进程启动时会等）； diff --git a/backend/app/cluster/version.py b/backend/app/cluster/version.py index c4450b6c..be3df40b 100644 --- a/backend/app/cluster/version.py +++ b/backend/app/cluster/version.py @@ -4,7 +4,8 @@ 规矩（认领、租约、接手的说法），两边各按各的写同一批行。以前只在 README 里写「升级要先停掉全部进程」，现在代码里拦住。 - 代码版本 code_version：包版本 + 构建号（AGENTLAB_BUILD，Docker 镜像构建时写进去）或 git 提交 + 源码指纹（app/ 下全部 .py - 的内容哈希）。比的是源码指纹：同一份代码一定相同，改了一行就不同，取不到 git、没有构建号时也区分得开。 + 的内容，加上依赖锁文件 requirements.lock 的内容；Docker 镜像里也留着这个文件）。比的是源码指纹：同一份代码、同一套依赖一定相同， + 改了一行、升了一个依赖就不同，取不到 git、没有构建号时也区分得开。锁文件取不到（不是从仓库或镜像装的）时换成几个关键包的版本。 - 库结构版本 schema_version：模型里各表各列（名字、类型、可空）加迁移清单的哈希。 怎么拦：PostgreSQL 时，补列之前先看一眼（preflight）——还有活着的、版本不同的进程就不启动；它的心跳停着（多半是强杀之后留下的、 @@ -33,6 +34,30 @@ class VersionMismatch(RuntimeError): """库里还有活着的、版本不同的进程。""" +#: 依赖锁文件（仓库里 backend/requirements.lock；Docker 镜像里 /app/backend/requirements.lock） +LOCK_FILE = _APP_DIR.parent / "requirements.lock" +#: 锁文件取不到时拿这几个包的版本代替（升级时最容易带出行为差别的那几个） +KEY_PACKAGES = ("fastapi", "pydantic", "sqlalchemy", "asyncpg", "psycopg", "langgraph", "langgraph-checkpoint", + "langgraph-checkpoint-postgres", "langchain-core", "starlette", "uvicorn") + + +def dependency_marker() -> bytes: + """依赖这一份进指纹的样子：锁文件的内容；取不到就是几个关键包的版本。""" + try: + return b"lock:" + LOCK_FILE.read_bytes() + except OSError: + pass + from importlib.metadata import PackageNotFoundError, version + + parts = [] + for name in KEY_PACKAGES: + try: + parts.append(f"{name}=={version(name)}") + except PackageNotFoundError: + parts.append(f"{name}==") + return ("packages:" + ";".join(parts)).encode() + + @lru_cache(maxsize=1) def source_fingerprint() -> str: digest = hashlib.sha256() @@ -44,6 +69,8 @@ def source_fingerprint() -> str: digest.update(path.read_bytes()) except OSError: continue + # 依赖升级（锁文件变了）也算版本不同：代码一行没改、换了一版 SQLAlchemy / LangGraph 的进程同样不许混进来（复核 R-b13） + digest.update(b"\0deps\0" + dependency_marker()) return digest.hexdigest()[:12] @@ -115,10 +142,13 @@ async def _peers(conn: Any) -> list[dict[str, Any]] | None: return [dict(r) for r in rows] -def _describe(peer: dict[str, Any]) -> str: +def _describe(peer: dict[str, Any], beating: set[str] | None = None) -> str: code = peer.get("code_version") or "（升级前的版本，没有记版本）" schema = peer.get("schema_version") or "未知" - return f"{identity.describe(peer['id'])}（代码 {code}，库结构 {schema}）" + state = "" + if beating is not None: + state = "，心跳在走" if peer["id"] in beating else "，心跳停着、等租约过期" + return f"{identity.describe(peer['id'])}（代码 {code}，库结构 {schema}{state}）" async def preflight(*, observe_s: float | None = None) -> None: @@ -141,11 +171,12 @@ async def preflight(*, observe_s: float | None = None) -> None: return for peer in mismatched: first_beats.setdefault(peer["id"], peer.get("heartbeat_at")) - # 心跳还在走：真活着，照实报错。心跳停着的（强杀、断电留下的登记）等它们的租约过期 - beating = [p for p in mismatched if p.get("heartbeat_at") != first_beats.get(p["id"])] + # 心跳还在走：真活着，照实报错。心跳停着的（强杀、断电留下的登记）等它们的租约过期。报错时列出全部活着的、版本不同的 + # 进程（复核 R-b13：只点名心跳在走的那个，人停掉它再起又会被另一个拦下） + beating = {p["id"] for p in mismatched if p.get("heartbeat_at") != first_beats.get(p["id"])} waited = loop.time() - started if beating or waited >= lease_s() * 2 + observe: - raise VersionMismatch(_message(beating or mismatched)) + raise VersionMismatch(_message(mismatched, beating=beating)) if waited >= observe and not warned: warned = True logger.warning("库里登记着 %d 个版本不同的进程，心跳停着（多半是强杀之后留下的）：等它们的租约过期再启动", @@ -174,8 +205,8 @@ def _naive(moment: Any) -> Any: return moment.replace(tzinfo=None) if getattr(moment, "tzinfo", None) else moment -def _message(peers: list[dict[str, Any]]) -> str: - listed = "；".join(_describe(p) for p in peers) +def _message(peers: list[dict[str, Any]], *, beating: set[str] | None = None) -> str: + listed = "；".join(_describe(p, beating) for p in peers) return (f"库里还有 {len(peers)} 个版本不同的 AgentLab 进程活着：{listed}。本进程是代码 {code_version()}、库结构 " f"{schema_version()}。不支持新旧版本同时运行：先停掉这些进程再启动新版本（升级步骤见 README「多进程」的「升级」）") diff --git a/backend/app/db/base.py b/backend/app/db/base.py index b76ba890..f3bf306b 100644 --- a/backend/app/db/base.py +++ b/backend/app/db/base.py @@ -600,10 +600,19 @@ def _migrate(conn) -> None: # pragma: no cover - 同步回调 _backfill_not_null(conn, columns) +#: 存量库里 run_events 撞了号、这一次没建唯一索引时记下的结果（settings 表里；cluster: 开头的不是用户设置，api/settings 不列） +SEQ_DUPLICATES_KEY = "cluster:run_events_seq_duplicates" +#: 记下的撞号最多几处（之后启动只核对这几处还在不在） +SEQ_DUPLICATES_KEPT = 200 + + 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）。存量库里可能已经 有撞了号的（多进程之前的进程内计数、按需裁判的老写法）：先数一数，有就照实记日志、这一次不建（照旧用不唯一的索引），不让启动 - 失败——那几次运行的封存核对本来就对不上，删哪一条都不对，要人看。""" + 失败——那几次运行的封存核对本来就对不上，删哪一条都不对，要人看。 + + 数一次就记下结果（settings 表的 SEQ_DUPLICATES_KEY：撞了号的那几处），之后启动不再整表 GROUP BY（复核 R-b13）：只按索引核对 + 记下的那几处还撞不撞，还撞着就一句话带过；都清掉了才再整表数一遍、建索引、删掉这条记录。""" from sqlalchemy import inspect, text if "run_events" not in columns: @@ -611,16 +620,76 @@ def _unique_run_event_seq(conn: Any, columns: dict[str, set[str]]) -> None: # p 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() + known = _known_seq_duplicates(conn, columns) + if known: + still = _still_duplicated(conn, known["pairs"]) + if still: + logger.info("run_events 里还有撞了号的事件（%s 记下的 %d 处里还剩 %d 处）：(run_id, seq) 的唯一索引照旧不建，" + "处理掉之后下次启动会再核对一遍", known.get("found_at") or "之前", len(known["pairs"]), still) + return + dupes = conn.execute(text("SELECT run_id, seq FROM run_events GROUP BY run_id, seq HAVING COUNT(*) > 1 LIMIT :n"), + {"n": SEQ_DUPLICATES_KEPT + 1}).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])) + logger.warning("run_events 里有撞了号的事件（%d 处%s，涉及运行 %s）：这一次不建 (run_id, seq) 的唯一索引。" + "这几次运行的封存核对对不上，请人工核对后处理（之后启动只核对这几处，不再整表查）", + min(len(dupes), SEQ_DUPLICATES_KEPT), "以上" if len(dupes) > SEQ_DUPLICATES_KEPT else "", + "、".join(runs[:5])) + _remember_seq_duplicates(conn, [[str(r[0]), int(r[1])] for r in dupes[:SEQ_DUPLICATES_KEPT]], + more=len(dupes) > SEQ_DUPLICATES_KEPT) 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")) + if known is not None: + _forget_seq_duplicates(conn) + logger.info("run_events 里的撞号都处理掉了：补上了 (run_id, seq) 的唯一索引") + + +def _known_seq_duplicates(conn: Any, columns: dict[str, set[str]]) -> dict[str, Any] | None: # pragma: no cover + if "settings" not in columns: + return None + from sqlalchemy import select + + from app.db.models import Setting + + row = conn.execute(select(Setting.value).where(Setting.key == SEQ_DUPLICATES_KEY)).first() + value = row[0] if row else None + if not isinstance(value, dict) or not isinstance(value.get("pairs"), list): + return None + return value + + +def _still_duplicated(conn: Any, pairs: list[Any]) -> int: # pragma: no cover + """记下的那几处里还撞着的有几处（每处按 (run_id, seq) 索引查一次）。""" + from sqlalchemy import func, select + + from app.db.models import RunEvent + + left = 0 + for pair in pairs: + if not (isinstance(pair, list) and len(pair) == 2): + continue + n = conn.execute(select(func.count()).select_from(RunEvent).where( + RunEvent.run_id == str(pair[0]), RunEvent.seq == int(pair[1]))).scalar() + left += int((n or 0) > 1) + return left + + +def _remember_seq_duplicates(conn: Any, pairs: list[list[Any]], *, more: bool) -> None: # pragma: no cover + from app.db.models import Setting + + _forget_seq_duplicates(conn) + conn.execute(Setting.__table__.insert().values( + key=SEQ_DUPLICATES_KEY, value={"pairs": pairs, "more": more, "found_at": utcnow().isoformat(timespec="seconds")})) + + +def _forget_seq_duplicates(conn: Any) -> None: # pragma: no cover + from sqlalchemy import delete + + from app.db.models import Setting + + conn.execute(delete(Setting).where(Setting.key == SEQ_DUPLICATES_KEY)) #: init_db 在 PostgreSQL 上拿的会话级咨询锁的号：几个进程同时启动时，建表和迁移一个一个来（同一个号别处别用） diff --git a/backend/tests/test_cluster.py b/backend/tests/test_cluster.py index ffe61b21..ece95be2 100644 --- a/backend/tests/test_cluster.py +++ b/backend/tests/test_cluster.py @@ -640,6 +640,109 @@ async def test_preflight_refuses_a_beating_peer_and_waits_out_a_stale_one(): await workers.register() +@pytest.mark.skipif(app_settings.db_dialect != "postgresql", reason="preflight 只在 PostgreSQL 上看") +async def test_preflight_names_every_live_process_of_another_version(): + """复核 R-b13：一个心跳在走、一个心跳停着——报错两个都列出来（只点名先看到心跳的那个，人停掉它再起还会被另一个拦下）。""" + from app.cluster import version + + await _peer_version(PEER, "0.0.9+old+aaaaaaaaaaaa") + await _peer_version(GHOST, "0.0.8+older+bbbbbbbbbbbb") + + async def beat() -> None: + for _ in range(15): + await asyncio.sleep(0.1) + async with SessionLocal() as session: + row = await session.get(ClusterWorker, PEER) + row.heartbeat_at = utcnow() + await session.commit() + + beating = asyncio.create_task(beat()) + with pytest.raises(version.VersionMismatch) as e: + await version.preflight(observe_s=0.5) + await beating + text = str(e.value) + assert "2 个版本不同" in text and "aaaaaaaaaaaa" in text and "bbbbbbbbbbbb" in text + assert "心跳在走" in text and "心跳停着" in text + + +def test_the_fingerprint_changes_with_the_dependency_lock(monkeypatch, tmp_path): + """复核 R-b13：依赖锁文件变了（升了依赖）源码指纹跟着变；锁文件取不到时换成几个关键包的版本。""" + from app.cluster import version + + lock = tmp_path / "requirements.lock" + monkeypatch.setattr(version, "LOCK_FILE", lock) + + def fingerprint() -> str: + version.source_fingerprint.cache_clear() + return version.source_fingerprint() + + try: + lock.write_text("sqlalchemy==2.0.40\n") + first = fingerprint() + assert fingerprint() == first + lock.write_text("sqlalchemy==2.0.41\n") + assert fingerprint() != first, "升了一个依赖：指纹不同" + lock.unlink() + assert version.dependency_marker().startswith(b"packages:") and b"sqlalchemy==" in version.dependency_marker() + assert fingerprint() not in (first,) + finally: + version.source_fingerprint.cache_clear() + + +async def test_a_known_seq_collision_is_not_rescanned_on_every_start(tmp_path, caplog): + """复核 R-b13：存量库里有撞了号的事件——第一次启动整表数一遍、记下结果、不建唯一索引；之后启动只核对记下的那几处，不再整表 + GROUP BY；撞号处理掉之后再数一遍、建索引、删掉记录。用一个单独的 SQLite 库（两种库的测试里都跑）。""" + import logging + + from sqlalchemy import event, inspect, text + + from app.db import base as db_base + from app.db.models import Setting + + target = db_base.make_engine(f"sqlite+aiosqlite:///{tmp_path / 'seq.db'}") + scans: list[str] = [] + event.listen(target.sync_engine, "before_cursor_execute", + lambda conn, cursor, statement, *a: scans.append(statement) if "GROUP BY run_id, seq" in statement else None) + try: + await db_base.create_schema(target) + from sqlalchemy.ext.asyncio import async_sessionmaker + + async with target.begin() as conn: + await conn.execute(text("DROP INDEX ux_run_events_run_seq")) + await conn.execute(text("CREATE INDEX ix_run_events_run_seq ON run_events (run_id, seq)")) + async with async_sessionmaker(target, expire_on_commit=False)() as session: + session.add(Run(id="r1", status="succeeded", graph={}, input={})) + await session.flush() + session.add_all([RunEvent(id=eid, run_id="r1", seq=seq, type="log", ts=0, data={}) + for eid, seq in (("e1", 1), ("e2", 1), ("e3", 2))]) + await session.commit() + + async def start() -> set[str]: + async with target.begin() as conn: + def run(sync_conn): + cols = {t: {c["name"] for c in inspect(sync_conn).get_columns(t)} for t in ("run_events", "settings")} + db_base._unique_run_event_seq(sync_conn, cols) + return {ix["name"] for ix in inspect(sync_conn).get_indexes("run_events")} + return await conn.run_sync(run) + + caplog.set_level(logging.INFO, logger="app.db.base") + assert "ux_run_events_run_seq" not in await start() and len(scans) == 1 + async with target.connect() as conn: + marker = (await conn.execute(text(f"SELECT value FROM settings WHERE key = '{db_base.SEQ_DUPLICATES_KEY}'"))).scalar() + assert '"r1"' in str(marker) and "1" in str(marker) + assert "ux_run_events_run_seq" not in await start() and len(scans) == 1, "第二次启动不再整表 GROUP BY" + assert any("还有撞了号的事件" in r.getMessage() for r in caplog.records) + async with target.begin() as conn: + await conn.execute(text("DELETE FROM run_events WHERE id = 'e2'")) + assert "ux_run_events_run_seq" in await start() and len(scans) == 2, "处理掉之后再数一遍、建上索引" + async with target.connect() as conn: + left = (await conn.execute(text(f"SELECT COUNT(*) FROM settings WHERE key = '{db_base.SEQ_DUPLICATES_KEY}'"))).scalar() + assert left == 0 + assert Setting.__tablename__ == "settings" + finally: + await target.dispose() + + @pytest.mark.sqlite_only async def test_an_idle_single_process_writes_nothing_for_coordination(monkeypatch): """复核 R-b3：SQLite（只有一个进程）空闲时协调不写库——不续租约、不写当班、不周期接手；要名额、放名额也不落库。用另一条连接